1use crate::dictionary::Dictionary;
18use crate::header::{
19 Header, FLAG_HAS_QUADS, FLAG_HAS_QUOTED_TRIPLES, FLAG_TILE_SYNOPSIS, HEADER_LEN, MAGIC,
20};
21use crate::index::{GraphIndex, IndexPermutation, Pattern, PermSet, NUM_PERMS};
22use crate::meta::{ClassNode, CommunityDescriptor, LevelLinks, LevelRollup, PyramidMeta};
23use crate::pyramid::{build_dendrogram, project_graph, PyramidAlgo};
24use crate::reader::RangeReader;
25use crate::tiling::{choose_round_for_budget, summarize, SuperEdge};
26use crate::triples::Triple;
27use crate::varint::{read_uvarint, write_uvarint};
28
29pub const DEFAULT_TILE_BUDGET: usize = 64 * 1024;
31
32pub fn build_pyramid_meta(
41 dict: &Dictionary,
42 triples: &[(u32, u32, u32)],
43 budget: usize,
44) -> (Vec<u8>, u16) {
45 build_pyramid_meta_with(dict, triples, budget, None)
46}
47
48pub fn build_pyramid_meta_with(
52 dict: &Dictionary,
53 triples: &[(u32, u32, u32)],
54 budget: usize,
55 type_override: Option<&str>,
56) -> (Vec<u8>, u16) {
57 build_pyramid_meta_algo(dict, triples, budget, type_override, PyramidAlgo::Louvain)
58}
59
60pub fn build_pyramid_meta_algo(
67 dict: &Dictionary,
68 triples: &[(u32, u32, u32)],
69 budget: usize,
70 type_override: Option<&str>,
71 algo: PyramidAlgo,
72) -> (Vec<u8>, u16) {
73 let timing = std::env::var_os("RETE_BUILD_TIMING").is_some();
79 let mut t = timing.then(std::time::Instant::now);
80 let mut lap = |label: &str| {
81 if let Some(t0) = &mut t {
82 eprintln!(
83 " [pyramid] {label}: {:.0} ms",
84 t0.elapsed().as_secs_f64() * 1000.0
85 );
86 *t0 = std::time::Instant::now();
87 }
88 };
89
90 let louvain = |lap: &mut dyn FnMut(&str)| {
92 let g = project_graph(dict, triples);
93 lap("project_graph");
94 let d = build_dendrogram(&g);
95 lap("build_dendrogram (Louvain)");
96 d
97 };
98 let dend = match algo {
99 PyramidAlgo::Louvain => louvain(&mut lap),
100 PyramidAlgo::Types => {
101 match crate::schema_pyramid::build_type_dendrogram(dict, triples, type_override) {
102 Some(d) => {
103 lap("build_type_dendrogram");
104 d
105 }
106 None => {
107 eprintln!(
108 " [pyramid] --pyramid-algo types: no usable rdf:type \
109 predicate — falling back to louvain"
110 );
111 louvain(&mut lap)
112 }
113 }
114 }
115 };
116 let round = choose_round_for_budget(dict, triples, &dend, budget);
117 lap("choose_round_for_budget");
118 let summary = summarize(dict, triples, &dend, round);
119 lap("summarize");
120 let sp = crate::schema_pyramid::build_schema_pyramid_with(
125 dict,
126 triples,
127 &dend,
128 round,
129 type_override,
130 );
131 lap("build_schema_pyramid");
132 let predicate_stats = compute_predicate_stats(triples);
133 lap("compute_predicate_stats");
134 let char_sets = compute_char_sets(triples);
135 lap("compute_char_sets");
136 let label_index = compute_label_index(dict, triples);
137 lap("compute_label_index");
138 let meta = PyramidMeta::new(round as u32, summary, &[])
139 .with_schema(
140 sp.class_hierarchy,
141 sp.level_rollups,
142 sp.level_links,
143 sp.descriptors,
144 sp.subclass_cycles,
145 sp.disjoint_pairs,
146 sp.equivalent_pairs,
147 )
148 .with_predicate_stats(predicate_stats)
149 .with_char_sets(char_sets)
150 .with_label_index(label_index);
151 let out = (meta.encode(), dend.rounds() as u16);
152 lap("encode");
153 out
154}
155
156const LABEL_PREDICATES: &[&str] = &[
160 "<http://www.w3.org/2000/01/rdf-schema#label>",
161 "<http://www.w3.org/2004/02/skos/core#prefLabel>",
162 "<http://www.w3.org/2004/02/skos/core#altLabel>",
163 "<http://xmlns.com/foaf/0.1/name>",
164 "<http://purl.org/dc/terms/title>",
165 "<http://purl.org/dc/elements/1.1/title>",
166 "<http://schema.org/name>",
167];
168
169fn compute_label_index(
176 dict: &Dictionary,
177 triples: &[(u32, u32, u32)],
178) -> Vec<crate::meta::LabelEntry> {
179 use crate::terms::{is_literal, literal_lexical};
180 use std::collections::{HashMap, HashSet};
181 const MAX_LABELS: usize = 8192;
182
183 let label_pids: HashSet<u32> = LABEL_PREDICATES
185 .iter()
186 .filter_map(|p| dict.predicate_id(p))
187 .collect();
188 if label_pids.is_empty() {
189 return Vec::new();
190 }
191 let mut degree: HashMap<u32, u32> = HashMap::new();
193 for &(s, _p, _o) in triples {
194 *degree.entry(s).or_insert(0) += 1;
195 }
196 let mut seen: HashSet<(u32, String)> = HashSet::new();
198 let mut candidates: Vec<(u32, String, u32)> = Vec::new(); for &(s, p, o) in triples {
200 if !label_pids.contains(&p) {
201 continue;
202 }
203 let Some(term) = dict.object_term(o) else {
204 continue;
205 };
206 if !is_literal(&term) {
207 continue;
208 }
209 let Some(label) = literal_lexical(&term) else {
210 continue;
211 };
212 if label.is_empty() {
213 continue;
214 }
215 if seen.insert((s, label.to_lowercase())) {
216 candidates.push((*degree.get(&s).unwrap_or(&0), label, s));
217 }
218 }
219 if candidates.len() > MAX_LABELS {
222 candidates.sort_by(|a, b| {
223 b.0.cmp(&a.0)
224 .then_with(|| a.2.cmp(&b.2))
225 .then_with(|| a.1.cmp(&b.1))
226 });
227 candidates.truncate(MAX_LABELS);
228 }
229 candidates.sort_by(|a, b| {
231 a.1.to_lowercase()
232 .cmp(&b.1.to_lowercase())
233 .then_with(|| a.1.cmp(&b.1))
234 .then_with(|| a.2.cmp(&b.2))
235 });
236 candidates
237 .into_iter()
238 .map(|(_deg, label, subject)| crate::meta::LabelEntry { label, subject })
239 .collect()
240}
241
242pub(crate) fn compute_text_index(dict: &Dictionary, triples: &[(u32, u32, u32)]) -> Vec<u8> {
248 use crate::terms::{is_literal, literal_lexical};
249 let mut b = crate::text_index::TextIndexBuilder::new();
250 for &(s, _p, o) in triples {
251 let Some(term) = dict.object_term(o) else {
252 continue;
253 };
254 if !is_literal(&term) {
255 continue;
256 }
257 if let Some(lit) = literal_lexical(&term) {
258 b.add_text(&lit, s);
259 }
260 }
261 if b.is_empty() {
262 Vec::new()
263 } else {
264 b.build(writer_codec())
265 }
266}
267
268fn compute_char_sets(triples: &[(u32, u32, u32)]) -> Vec<crate::meta::CharSet> {
273 use std::collections::{BTreeSet, HashMap};
274 const MAX_CHAR_SETS: usize = 128;
275 let mut by_subject: HashMap<u32, BTreeSet<u32>> = HashMap::new();
276 for &(s, p, _o) in triples {
277 by_subject.entry(s).or_default().insert(p);
278 }
279 let mut shapes: HashMap<Vec<u32>, u64> = HashMap::new();
280 for set in by_subject.into_values() {
281 *shapes.entry(set.into_iter().collect()).or_insert(0) += 1;
282 }
283 let mut v: Vec<crate::meta::CharSet> = shapes
284 .into_iter()
285 .map(|(predicates, subjects)| crate::meta::CharSet {
286 predicates,
287 subjects,
288 })
289 .collect();
290 v.sort_by(|a, b| {
291 b.subjects
292 .cmp(&a.subjects)
293 .then_with(|| a.predicates.cmp(&b.predicates))
294 });
295 v.truncate(MAX_CHAR_SETS);
296 v
297}
298
299fn compute_predicate_stats(triples: &[(u32, u32, u32)]) -> Vec<crate::meta::PredStat> {
305 use std::collections::HashMap;
306 #[allow(clippy::type_complexity)]
307 let mut acc: HashMap<u32, (HashMap<u32, u32>, HashMap<u32, u32>, u64)> = HashMap::new();
308 for &(s, p, o) in triples {
309 let e = acc.entry(p).or_default();
310 *e.0.entry(s).or_insert(0) += 1;
311 *e.1.entry(o).or_insert(0) += 1;
312 e.2 += 1;
313 }
314 let mut stats: Vec<crate::meta::PredStat> = acc
315 .into_iter()
316 .map(|(predicate, (subj, obj, count))| crate::meta::PredStat {
317 predicate,
318 count,
319 distinct_subjects: subj.len() as u64,
320 distinct_objects: obj.len() as u64,
321 max_objects_per_subject: subj.values().copied().max().unwrap_or(0),
322 max_subjects_per_object: obj.values().copied().max().unwrap_or(0),
323 })
324 .collect();
325 stats.sort_by_key(|p| p.predicate);
326 stats
327}
328
329pub const CODEC_NONE: u8 = 0;
331pub const CODEC_ZSTD: u8 = 1;
333#[cfg(feature = "compression")]
335const ZSTD_LEVEL: i32 = 9;
336
337#[derive(Debug, thiserror::Error)]
338#[non_exhaustive]
339pub enum FileError {
340 #[error("header: {0}")]
341 Header(#[from] crate::header::HeaderError),
342 #[error("malformed container: {0}")]
343 Container(&'static str),
344 #[error("unknown codec: {0}")]
345 UnknownCodec(u8),
346 #[error("decompression failed: {0}")]
347 Decompress(std::io::Error),
348 #[error("io: {0}")]
349 Io(#[from] std::io::Error),
350}
351
352pub(crate) fn writer_codec() -> u8 {
355 if cfg!(feature = "compression") {
356 CODEC_ZSTD
357 } else {
358 CODEC_NONE
359 }
360}
361
362fn intersect_sorted(a: &[u32], b: &[u32]) -> Vec<u32> {
365 let mut out = Vec::with_capacity(a.len().min(b.len()));
366 let (mut i, mut j) = (0, 0);
367 while i < a.len() && j < b.len() {
368 match a[i].cmp(&b[j]) {
369 std::cmp::Ordering::Less => i += 1,
370 std::cmp::Ordering::Greater => j += 1,
371 std::cmp::Ordering::Equal => {
372 out.push(a[i]);
373 i += 1;
374 j += 1;
375 }
376 }
377 }
378 out
379}
380
381pub(crate) fn compress(codec: u8, bytes: &[u8]) -> Vec<u8> {
382 match codec {
383 #[cfg(feature = "compression")]
384 CODEC_ZSTD => {
385 zstd::encode_all(bytes, ZSTD_LEVEL).expect("zstd encode is infallible in-memory")
386 }
387 _ => bytes.to_vec(),
388 }
389}
390
391pub(crate) fn decompress(codec: u8, bytes: &[u8]) -> Result<Vec<u8>, FileError> {
392 match codec {
393 CODEC_NONE => Ok(bytes.to_vec()),
394 #[cfg(feature = "compression")]
403 CODEC_ZSTD => zstd::decode_all(bytes).map_err(FileError::Decompress),
404 #[cfg(not(feature = "compression"))]
409 CODEC_ZSTD => {
410 use std::io::Read;
411 let mut dec = ruzstd::decoding::StreamingDecoder::new(bytes)
412 .map_err(|e| FileError::Decompress(std::io::Error::other(e.to_string())))?;
413 let mut out = Vec::new();
414 dec.read_to_end(&mut out).map_err(FileError::Decompress)?;
415 Ok(out)
416 }
417 other => Err(FileError::UnknownCodec(other)),
418 }
419}
420
421const TILE_COALESCE_GAP: u64 = 4096;
425
426const DICT_COALESCE_GAP: u64 = 64 * 1024;
432
433pub const DEFAULT_EXPORT_BUDGET_MB: u64 = 4096;
440
441pub const DEFAULT_QUERY_BUDGET_MB: u64 = DEFAULT_EXPORT_BUDGET_MB;
450
451pub fn spill_threshold_for(budget: u64) -> u64 {
455 budget / 8
456}
457
458const MIN_DICT_CACHE_CAP: u64 = 64 << 20; const TRANSIENT_RESERVE_MAX: u64 = 128 << 20; #[derive(Debug, Clone, Copy)]
479pub struct MemoryBudget {
480 pub block_cache: u64,
482 pub dict_cache: u64,
484 pub tile_cache: u64,
486}
487
488pub fn split_memory_budget(total: u64) -> MemoryBudget {
499 if total == u64::MAX {
500 return MemoryBudget {
501 block_cache: u64::MAX,
502 dict_cache: u64::MAX,
503 tile_cache: u64::MAX,
504 };
505 }
506 let block_cache = (total / 8).min(crate::block_cache::DEFAULT_CACHE_CAP);
509 let tile_cache = (total / 16).min(256 << 20);
512 let reserve = (total / 32).min(TRANSIENT_RESERVE_MAX);
518 let dict_cache = total
521 .saturating_sub(block_cache)
522 .saturating_sub(tile_cache)
523 .saturating_sub(reserve)
524 .max(MIN_DICT_CACHE_CAP);
525 MemoryBudget {
526 block_cache,
527 dict_cache,
528 tile_cache,
529 }
530}
531
532#[derive(Debug, Clone, Copy, Default)]
535pub struct FixedOverhead {
536 pub dict_directory: u64,
538 pub index_directory: u64,
540 pub named_graphs: u64,
542}
543
544impl FixedOverhead {
545 pub fn total(&self) -> u64 {
547 self.dict_directory + self.index_directory + self.named_graphs
548 }
549}
550
551fn fmt_cap(cap: u64) -> String {
553 if cap == u64::MAX {
554 "unlimited".to_string()
555 } else {
556 format!("{}MiB", cap >> 20)
557 }
558}
559
560fn read_coalesced<R: RangeReader + ?Sized>(
566 reader: &R,
567 ranges: &[ByteRange],
568 gap: u64,
569) -> Option<Vec<Vec<u8>>> {
570 let mut spans: Vec<(u64, u64)> = Vec::new();
573 let mut span_of: Vec<usize> = Vec::with_capacity(ranges.len());
574 let mut i = 0;
575 while i < ranges.len() {
576 let start = ranges[i].offset;
577 let mut end = ranges[i].offset.checked_add(ranges[i].len)?;
578 let mut j = i + 1;
579 while j < ranges.len() {
580 let r = &ranges[j];
581 if r.offset < end || r.offset - end > gap {
582 break;
583 }
584 end = r.offset.checked_add(r.len)?;
585 j += 1;
586 }
587 let si = spans.len();
588 spans.push((start, end - start));
589 for _ in i..j {
590 span_of.push(si);
591 }
592 i = j;
593 }
594 let blobs = reader.read_many(&spans).ok()?;
595 if blobs.len() != spans.len() {
596 return None;
597 }
598 let mut out = Vec::with_capacity(ranges.len());
599 for (k, r) in ranges.iter().enumerate() {
600 let (span_start, _) = spans[span_of[k]];
601 let blob = &blobs[span_of[k]];
602 let lo = (r.offset - span_start) as usize;
603 let hi = lo.checked_add(r.len as usize)?;
604 out.push(blob.get(lo..hi)?.to_vec());
605 }
606 Some(out)
607}
608
609fn content_hash(parts: &[&[u8]]) -> [u8; 16] {
612 let mut h = blake3::Hasher::new();
613 for p in parts {
614 h.update(p);
615 }
616 let mut out = [0u8; 16];
617 out.copy_from_slice(&h.finalize().as_bytes()[..16]);
618 out
619}
620
621pub type TermTriple = (String, String, String);
623
624#[derive(Debug, Clone, Copy, PartialEq, Eq)]
627pub struct DumpPlan {
628 pub scan: Option<crate::index::ScanPlan>,
632 pub dictionary_bytes: u64,
641}
642
643#[derive(Debug, Clone)]
647pub struct LayoutSegment {
648 pub kind: &'static str,
649 pub label: String,
650 pub offset: u64,
651 pub len: u64,
652}
653
654#[derive(Debug, Clone, Copy, PartialEq, Eq)]
656pub struct ByteRange {
657 pub offset: u64,
658 pub len: u64,
659}
660
661impl ByteRange {
662 pub fn end(self) -> u64 {
663 self.offset + self.len
664 }
665}
666
667#[derive(Debug, Clone, PartialEq, Eq)]
669pub struct TripleProvenance {
670 pub terms: TermTriple,
672 pub ids: Triple,
674 pub graph: Option<String>,
676 pub matched_pattern: Pattern,
678 pub index_permutation: IndexPermutation,
680 pub dictionary_range: ByteRange,
682 pub index_range: ByteRange,
684 pub index_section_range: ByteRange,
687 pub pyramid_range: Option<ByteRange>,
689 pub tile: Option<String>,
693 pub tile_range: Option<ByteRange>,
696}
697
698fn encode_container(sections: &[&[u8]], codec: u8) -> Vec<u8> {
702 let mut out = Vec::new();
703 write_uvarint(&mut out, sections.len() as u64);
704 for s in sections {
705 let payload = compress(codec, s);
706 write_uvarint(&mut out, payload.len() as u64);
707 out.extend_from_slice(&payload);
708 }
709 out
710}
711
712fn decode_container(bytes: &[u8], codec: u8) -> Result<Vec<Vec<u8>>, FileError> {
714 let (n, mut pos) = read_uvarint(bytes).ok_or(FileError::Container("truncated count"))?;
715 let mut out = Vec::with_capacity((n as usize).min(bytes.len()));
719 for _ in 0..n {
720 let (len, used) =
721 read_uvarint(&bytes[pos..]).ok_or(FileError::Container("truncated length"))?;
722 pos += used;
723 let end = pos + len as usize;
724 if end > bytes.len() {
725 return Err(FileError::Container("section overruns buffer"));
726 }
727 out.push(decompress(codec, &bytes[pos..end])?);
728 pos = end;
729 }
730 Ok(out)
731}
732
733fn checked_end(off: u64, len: u64) -> Result<u64, FileError> {
734 off.checked_add(len)
735 .ok_or(FileError::Container("section range overflows"))
736}
737
738const DICT_CHUNK_BUDGET: usize = 64 * 1024;
741
742fn encode_chunked_dict_section(raw: &[u8], codec: u8) -> Vec<u8> {
759 let meta = crate::dict::parse_meta(raw).unwrap_or(crate::dict::SectionMeta {
760 term_count: 0,
761 restart_interval: 1,
762 restart_offsets: Vec::new(),
763 });
764 let body_start = meta
765 .restart_offsets
766 .first()
767 .copied()
768 .unwrap_or(raw.len() as u64);
769 let header = &raw[..(body_start.min(raw.len() as u64)) as usize];
770
771 let n_runs = meta.restart_offsets.len();
773 let mut bounds: Vec<(usize, u64, u64)> = Vec::new(); let mut r = 0;
775 while r < n_runs {
776 let start = meta.restart_offsets[r];
777 let mut r2 = r + 1;
778 while r2 < n_runs && meta.restart_offsets[r2] - start < DICT_CHUNK_BUDGET as u64 {
779 r2 += 1;
780 }
781 let end = if r2 < n_runs {
782 meta.restart_offsets[r2]
783 } else {
784 raw.len() as u64
785 };
786 bounds.push((r, start, end));
787 r = r2;
788 }
789
790 let compressed: Vec<Vec<u8>> = bounds
791 .iter()
792 .map(|&(_, s, e)| compress(codec, &raw[s as usize..e as usize]))
793 .collect();
794 let mut out = Vec::new();
795 write_uvarint(&mut out, header.len() as u64);
796 out.extend_from_slice(header);
797 write_uvarint(&mut out, bounds.len() as u64);
798 let mut prev_run = 0usize;
799 let mut prev_last: Option<Vec<u8>> = None;
803 for (i, (&(first_run, start, end), comp)) in bounds.iter().zip(&compressed).enumerate() {
804 let key = if i == 0 {
805 Vec::new()
807 } else {
808 let first = crate::dict::run_first_term(raw, start as usize).unwrap_or_default();
809 match &prev_last {
810 None => first,
813 Some(pl) => crate::dict::shortest_separator(pl, &first),
814 }
815 };
816 if i + 1 < bounds.len() {
817 let last_run_off = meta.restart_offsets[bounds[i + 1].0 - 1] as usize;
819 prev_last = crate::dict::run_last_term(raw, last_run_off, end as usize);
820 }
821 write_uvarint(&mut out, (first_run - prev_run) as u64);
822 write_uvarint(&mut out, key.len() as u64);
823 out.extend_from_slice(&key);
824 write_uvarint(&mut out, comp.len() as u64);
825 prev_run = first_run;
826 }
827 for comp in &compressed {
828 out.extend_from_slice(comp);
829 }
830 out
831}
832
833struct DictChunkEntry {
837 first_run: usize,
838 key: Vec<u8>,
839 body_start: u64,
840 start: u64,
841 end: u64,
842}
843
844fn parse_chunked_dict_dir(
848 bytes: &[u8],
849 total_len: u64,
850) -> Result<(crate::dict::SectionMeta, Vec<DictChunkEntry>), FileError> {
851 let mut pos = 0usize;
852 let take = |pos: &mut usize| -> Result<u64, FileError> {
853 let (v, n) = read_uvarint(bytes.get(*pos..).unwrap_or(&[]))
854 .ok_or(FileError::Container("truncated dict chunk directory"))?;
855 *pos += n;
856 Ok(v)
857 };
858 let header_len = take(&mut pos)? as usize;
859 let header = bytes
860 .get(pos..pos.saturating_add(header_len))
861 .ok_or(FileError::Container("truncated dict header"))?;
862 let meta = crate::dict::parse_meta(header)
863 .map_err(|_| FileError::Container("malformed dict header"))?;
864 pos += header_len;
865
866 let num_chunks = take(&mut pos)? as usize;
867 let mut entries = Vec::with_capacity(num_chunks.min(bytes.len()));
868 let mut lens = Vec::with_capacity(num_chunks.min(bytes.len()));
869 let mut prev_run = 0usize;
870 for _ in 0..num_chunks {
871 let drun = take(&mut pos)? as usize;
872 let klen = take(&mut pos)? as usize;
873 let key = bytes
874 .get(pos..pos.saturating_add(klen))
875 .ok_or(FileError::Container("truncated dict chunk key"))?
876 .to_vec();
877 pos += klen;
878 let clen = take(&mut pos)?;
879 let first_run = prev_run + drun;
880 let body_start = meta
881 .restart_offsets
882 .get(first_run)
883 .copied()
884 .ok_or(FileError::Container("dict chunk run out of range"))?;
885 entries.push(DictChunkEntry {
886 first_run,
887 key,
888 body_start,
889 start: 0,
890 end: 0,
891 });
892 lens.push(clen);
893 prev_run = first_run;
894 }
895 let mut start = pos as u64;
896 for (e, len) in entries.iter_mut().zip(lens) {
897 let end = start
898 .checked_add(len)
899 .filter(|&e| e <= total_len)
900 .ok_or(FileError::Container("dict chunk overruns section"))?;
901 e.start = start;
902 e.end = end;
903 start = end;
904 }
905 Ok((meta, entries))
906}
907
908fn read_dict_dir_ranged<R: RangeReader>(
920 reader: &R,
921 section: ByteRange,
922) -> Result<(crate::dict::SectionMeta, Vec<DictChunkEntry>), FileError> {
923 let total = section.len;
924 let init = 8192.min(total); let head = reader.read_at(section.offset, init)?;
932 let (header_len, n0) =
933 read_uvarint(&head).ok_or(FileError::Container("truncated dict header len"))?;
934 let hbase = n0; let (term_count, n1) = read_uvarint(head.get(hbase..).unwrap_or(&[]))
936 .ok_or(FileError::Container("truncated dict term_count"))?;
937 let (restart_interval, _n2) = read_uvarint(head.get(hbase + n1..).unwrap_or(&[]))
938 .ok_or(FileError::Container("truncated dict interval"))?;
939 if restart_interval == 0 {
940 return Err(FileError::Container("zero restart interval"));
941 }
942 let dir_start = (hbase as u64)
945 .checked_add(header_len)
946 .filter(|&d| d <= total)
947 .ok_or(FileError::Container("dict header overruns section"))?;
948 let dir_total = total - dir_start;
949 let meta = crate::dict::SectionMeta {
950 term_count: term_count as u32,
951 restart_interval: restart_interval as u32,
952 restart_offsets: Vec::new(),
953 };
954 let finish = |mut entries: Vec<DictChunkEntry>| {
955 for e in &mut entries {
956 e.start += dir_start; e.end += dir_start;
958 }
959 (meta.clone(), entries)
960 };
961 let mut have: Vec<u8> = Vec::new();
965 if dir_start < head.len() as u64 {
966 match parse_chunk_dir_only(&head[dir_start as usize..], dir_total)? {
967 ChunkDirParse::Done(entries) => return Ok(finish(entries)),
968 ChunkDirParse::Truncated { .. } => have = head[dir_start as usize..].to_vec(),
969 }
970 }
971 let mut want = 4096u64.min(dir_total).max(1);
984 loop {
985 let held = have.len() as u64;
986 if want > held {
987 let extra = reader.read_at(section.offset + dir_start + held, want - held)?;
988 if extra.is_empty() {
989 return Err(FileError::Container("truncated dict chunk directory"));
990 }
991 have.extend_from_slice(&extra);
992 }
993 let held = have.len() as u64;
994 match parse_chunk_dir_only(&have, dir_total)? {
995 ChunkDirParse::Done(entries) => return Ok(finish(entries)),
996 ChunkDirParse::Truncated { .. } if held >= dir_total => {
997 return Err(FileError::Container("truncated dict chunk directory"));
998 }
999 ChunkDirParse::Truncated {
1000 parsed,
1001 used,
1002 total,
1003 } => {
1004 let est = if parsed >= 16 && used > 0 {
1009 let whole = ((used as u64) / (parsed as u64)).saturating_mul(total as u64);
1010 whole.saturating_add(whole / 8).saturating_add(64)
1011 } else {
1012 held.saturating_mul(2)
1014 };
1015 want = est
1016 .max(held.saturating_add(1))
1017 .min(held.saturating_mul(4))
1018 .min(dir_total);
1019 }
1020 }
1021 }
1022}
1023
1024enum ChunkDirParse {
1029 Done(Vec<DictChunkEntry>),
1030 Truncated {
1031 parsed: usize,
1033 used: usize,
1035 total: usize,
1037 },
1038}
1039
1040fn parse_chunk_dir_only(dir: &[u8], dir_total: u64) -> Result<ChunkDirParse, FileError> {
1050 let mut pos = 0usize;
1051 let take = |pos: &mut usize| -> Option<u64> {
1052 let (v, n) = read_uvarint(dir.get(*pos..).unwrap_or(&[]))?;
1053 *pos += n;
1054 Some(v)
1055 };
1056 let Some(num_chunks) = take(&mut pos) else {
1057 return Ok(ChunkDirParse::Truncated {
1058 parsed: 0,
1059 used: 0,
1060 total: 0,
1061 });
1062 };
1063 let num_chunks = num_chunks as usize;
1064 let mut entries = Vec::with_capacity(num_chunks.min(dir.len()));
1065 let mut lens: Vec<u64> = Vec::with_capacity(num_chunks.min(dir.len()));
1066 let mut prev_run = 0usize;
1067 for _ in 0..num_chunks {
1068 let entry_start = pos;
1069 let short = ChunkDirParse::Truncated {
1070 parsed: entries.len(),
1071 used: entry_start,
1072 total: num_chunks,
1073 };
1074 let (Some(drun), Some(klen)) = (take(&mut pos), take(&mut pos)) else {
1075 return Ok(short);
1076 };
1077 let (drun, klen) = (drun as usize, klen as usize);
1078 let Some(key) = dir.get(pos..pos.saturating_add(klen)) else {
1079 return Ok(short);
1080 };
1081 let key = key.to_vec();
1082 pos += klen;
1083 let Some(clen) = take(&mut pos) else {
1084 return Ok(short);
1085 };
1086 let first_run = prev_run + drun;
1087 entries.push(DictChunkEntry {
1088 first_run,
1089 key,
1090 body_start: 0,
1091 start: 0,
1092 end: 0,
1093 });
1094 lens.push(clen);
1095 prev_run = first_run;
1096 }
1097 let mut start = pos as u64;
1098 for (e, len) in entries.iter_mut().zip(lens) {
1099 let end = start
1100 .checked_add(len)
1101 .filter(|&e| e <= dir_total)
1102 .ok_or(FileError::Container("dict chunk overruns section"))?;
1103 e.start = start;
1104 e.end = end;
1105 start = end;
1106 }
1107 Ok(ChunkDirParse::Done(entries))
1108}
1109
1110fn decode_chunked_dict_section(
1114 payload: &[u8],
1115 codec: u8,
1116 cache: std::sync::Arc<crate::chunk_cache::ChunkCache>,
1117 section_index: u8,
1118) -> Result<crate::dict::ChunkedSection, FileError> {
1119 let (meta, entries) = parse_chunked_dict_dir(payload, payload.len() as u64)?;
1120 let chunks = entries
1121 .into_iter()
1122 .map(|e| {
1123 let body = decompress(codec, &payload[e.start as usize..e.end as usize])?;
1124 Ok((
1125 crate::dict::SectionChunk::new(e.first_run, e.key, e.body_start),
1126 body,
1127 ))
1128 })
1129 .collect::<Result<Vec<_>, FileError>>()?;
1130 Ok(crate::dict::ChunkedSection::resident(
1131 meta,
1132 chunks,
1133 cache,
1134 section_index,
1135 ))
1136}
1137
1138fn decode_dictionary_container(bytes: &[u8], codec: u8) -> Result<Dictionary, FileError> {
1139 let dsecs = decode_container(bytes, CODEC_NONE)?;
1140 if dsecs.len() != 4 {
1141 return Err(FileError::Container("expected 4 dictionary sections"));
1142 }
1143 let cache = crate::chunk_cache::ChunkCache::unlimited_arc();
1146 let mut sections = Vec::with_capacity(4);
1147 for (si, sec) in dsecs.iter().enumerate() {
1148 sections.push(decode_chunked_dict_section(
1149 sec,
1150 codec,
1151 cache.clone(),
1152 si as u8,
1153 )?);
1154 }
1155 let arr: [crate::dict::ChunkedSection; 4] = sections
1156 .try_into()
1157 .map_err(|_| FileError::Container("expected 4 dictionary sections"))?;
1158 Ok(Dictionary::from_chunked_sections(arr))
1159}
1160
1161fn encode_tiled_section(index: &GraphIndex, si: usize, codec: u8) -> Vec<u8> {
1167 let tiles = index.tile_sections()[si];
1168 let n = tiles.len();
1169 #[cfg(feature = "parallel")]
1176 let compressed: Vec<Vec<u8>> = {
1177 use rayon::prelude::*;
1178 (0..n)
1179 .into_par_iter()
1180 .map(|ti| compress(codec, &index.tile_body(si, ti)))
1181 .collect()
1182 };
1183 #[cfg(not(feature = "parallel"))]
1184 let compressed: Vec<Vec<u8>> = (0..n)
1185 .map(|ti| compress(codec, &index.tile_body(si, ti)))
1186 .collect();
1187 let mut out = Vec::new();
1188 write_uvarint(&mut out, tiles.len() as u64);
1189 let mut prev_min = 0u32;
1190 for (tile, comp) in tiles.iter().zip(&compressed) {
1191 let (min_a, max_a) = tile.leading_range();
1192 write_uvarint(&mut out, (min_a - prev_min) as u64);
1193 write_uvarint(&mut out, (max_a - min_a) as u64);
1194 write_uvarint(&mut out, comp.len() as u64);
1195 prev_min = min_a;
1196 }
1197 for comp in &compressed {
1198 out.extend_from_slice(comp);
1199 }
1200 for ti in 0..n {
1208 let body = index.tile_body(si, ti);
1209 let (min_b, max_b, min_c, max_c) = match crate::triples::TripleBlock::parse(&body) {
1210 Ok(b) => {
1211 let z = b.zone();
1212 (z.min_b, z.max_b, z.min_c, z.max_c)
1213 }
1214 Err(_) => (0, u32::MAX, 0, u32::MAX),
1215 };
1216 write_uvarint(&mut out, min_b as u64);
1217 write_uvarint(&mut out, (max_b - min_b) as u64);
1218 write_uvarint(&mut out, min_c as u64);
1219 write_uvarint(&mut out, (max_c - min_c) as u64);
1220 }
1221 out
1222}
1223
1224struct TileDirEntry {
1227 min_a: u32,
1228 max_a: u32,
1229 start: u64,
1230 end: u64,
1231}
1232
1233type TileSynopsis = (u32, u32, u32, u32);
1235
1236fn parse_tile_synopsis(
1244 payload: &[u8],
1245 trailer_start: usize,
1246 num_tiles: usize,
1247) -> Option<Vec<TileSynopsis>> {
1248 let mut pos = trailer_start;
1249 let take = |pos: &mut usize| -> Option<u32> {
1250 let (v, n) = read_uvarint(payload.get(*pos..)?)?;
1251 *pos += n;
1252 u32::try_from(v).ok()
1253 };
1254 let mut out = Vec::with_capacity(num_tiles.min(payload.len()));
1255 for _ in 0..num_tiles {
1256 let min_b = take(&mut pos)?;
1257 let max_b = min_b.checked_add(take(&mut pos)?)?;
1258 let min_c = take(&mut pos)?;
1259 let max_c = min_c.checked_add(take(&mut pos)?)?;
1260 out.push((min_b, max_b, min_c, max_c));
1261 }
1262 Some(out)
1263}
1264
1265fn parse_tile_directory(bytes: &[u8], total_len: u64) -> Result<Vec<TileDirEntry>, FileError> {
1270 let mut pos = 0usize;
1271 let take = |pos: &mut usize| -> Result<u64, FileError> {
1272 let (v, n) = read_uvarint(bytes.get(*pos..).unwrap_or(&[]))
1273 .ok_or(FileError::Container("truncated tile directory"))?;
1274 *pos += n;
1275 Ok(v)
1276 };
1277 let num_tiles = take(&mut pos)? as usize;
1278 let mut entries = Vec::with_capacity(num_tiles.min(bytes.len()));
1279 let mut prev_min = 0u32;
1280 let mut lens = Vec::with_capacity(num_tiles.min(bytes.len()));
1281 for _ in 0..num_tiles {
1282 let dmin = take(&mut pos)? as u32;
1283 let span = take(&mut pos)? as u32;
1284 let len = take(&mut pos)?;
1285 let min_a = prev_min.wrapping_add(dmin);
1286 entries.push(TileDirEntry {
1287 min_a,
1288 max_a: min_a.wrapping_add(span),
1289 start: 0,
1290 end: 0,
1291 });
1292 lens.push(len);
1293 prev_min = min_a;
1294 }
1295 let mut start = pos as u64;
1296 for (e, len) in entries.iter_mut().zip(lens) {
1297 let end = start
1298 .checked_add(len)
1299 .filter(|&e| e <= total_len)
1300 .ok_or(FileError::Container("tile overruns section"))?;
1301 e.start = start;
1302 e.end = end;
1303 start = end;
1304 }
1305 Ok(entries)
1306}
1307
1308fn read_tile_directory_ranged<R: RangeReader>(
1312 reader: &R,
1313 section: ByteRange,
1314) -> Result<Vec<TileDirEntry>, FileError> {
1315 let total = section.len;
1316 let mut prefetch = 4096u64.min(total);
1317 loop {
1318 let prefix = reader.read_at(section.offset, prefetch)?;
1319 match parse_tile_directory(&prefix, total) {
1320 Ok(dir) => return Ok(dir),
1321 Err(_) if prefetch < total => prefetch = prefetch.saturating_mul(2).min(total),
1322 Err(e) => return Err(e),
1323 }
1324 }
1325}
1326
1327fn read_tile_synopsis_ranged<R: RangeReader>(
1334 reader: &R,
1335 section: ByteRange,
1336 dir: &[TileDirEntry],
1337) -> Vec<Option<TileSynopsis>> {
1338 let n = dir.len();
1339 let none = vec![None; n];
1340 let trailer_start = dir.iter().map(|e| e.end).max().unwrap_or(0);
1341 let total = section.len;
1342 if n == 0 || trailer_start >= total {
1343 return none; }
1345 let trailer_len = total - trailer_start;
1346 let Ok(bytes) = reader.read_at(section.offset + trailer_start, trailer_len) else {
1347 return none;
1348 };
1349 match parse_tile_synopsis(&bytes, 0, n) {
1350 Some(v) => v.into_iter().map(Some).collect(),
1351 None => none,
1352 }
1353}
1354
1355fn tile_file_ranges(
1359 index_bytes: &[u8],
1360 container_offset: u64,
1361 section_ranges: &[ByteRange; NUM_PERMS],
1362) -> [Vec<(u32, u32, ByteRange)>; NUM_PERMS] {
1363 let mut out: [Vec<(u32, u32, ByteRange)>; NUM_PERMS] = Default::default();
1364 for (section, range) in out.iter_mut().zip(section_ranges) {
1365 if range.len == 0 || range.offset < container_offset {
1369 continue;
1370 }
1371 let start = (range.offset - container_offset) as usize;
1372 let Some(payload) = index_bytes.get(start..start + range.len as usize) else {
1373 continue;
1374 };
1375 if let Ok(dir) = parse_tile_directory(payload, payload.len() as u64) {
1376 *section = dir
1377 .into_iter()
1378 .map(|e| {
1379 (
1380 e.min_a,
1381 e.max_a,
1382 ByteRange {
1383 offset: range.offset + e.start,
1384 len: (e.end - e.start),
1385 },
1386 )
1387 })
1388 .collect();
1389 }
1390 }
1391 out
1392}
1393
1394fn decode_tiled_section(payload: &[u8], codec: u8) -> Result<Vec<(u32, u32, Vec<u8>)>, FileError> {
1397 parse_tile_directory(payload, payload.len() as u64)?
1398 .into_iter()
1399 .map(|e| {
1400 Ok((
1401 e.min_a,
1402 e.max_a,
1403 decompress(codec, &payload[e.start as usize..e.end as usize])?,
1404 ))
1405 })
1406 .collect()
1407}
1408
1409fn decode_index_container(
1419 bytes: &[u8],
1420 codec: u8,
1421 perms: PermSet,
1422) -> Result<GraphIndex, FileError> {
1423 let mut isecs = decode_container(bytes, CODEC_NONE)?;
1424 if isecs.len() != perms.len() {
1425 return Err(FileError::Container(
1426 "index container section count does not match the header permutation mask",
1427 ));
1428 }
1429 let mut sections: [Vec<(u32, u32, Vec<u8>)>; NUM_PERMS] = Default::default();
1430 for (perm, sec) in perms.iter().zip(isecs.iter_mut()) {
1431 sections[perm.section_index()] = decode_tiled_section(sec, codec)?;
1432 }
1433 Ok(GraphIndex::from_tiles(sections, perms))
1434}
1435
1436fn container_section_payload_ranges(
1437 bytes: &[u8],
1438 container_offset: u64,
1439 expected_sections: usize,
1440) -> Result<Vec<ByteRange>, FileError> {
1441 let (section_count, mut pos) =
1442 read_uvarint(bytes).ok_or(FileError::Container("truncated count"))?;
1443 let section_count = usize::try_from(section_count)
1444 .map_err(|_| FileError::Container("section count too large"))?;
1445 if section_count != expected_sections {
1446 return Err(FileError::Container("unexpected section count"));
1447 }
1448
1449 let mut ranges = Vec::with_capacity(section_count);
1450 for _ in 0..section_count {
1451 let remaining = bytes
1452 .get(pos..)
1453 .ok_or(FileError::Container("truncated length"))?;
1454 let (payload_len, used) =
1455 read_uvarint(remaining).ok_or(FileError::Container("truncated length"))?;
1456 pos = pos
1457 .checked_add(used)
1458 .ok_or(FileError::Container("section range overflows"))?;
1459 let payload_len_usize = usize::try_from(payload_len)
1460 .map_err(|_| FileError::Container("section length too large"))?;
1461 let payload_end = pos
1462 .checked_add(payload_len_usize)
1463 .ok_or(FileError::Container("section range overflows"))?;
1464 if payload_end > bytes.len() {
1465 return Err(FileError::Container("section overruns buffer"));
1466 }
1467 ranges.push(ByteRange {
1468 offset: checked_end(container_offset, pos as u64)?,
1469 len: payload_len,
1470 });
1471 pos = payload_end;
1472 }
1473
1474 Ok(ranges)
1475}
1476
1477fn decode_index_section_ranges(
1478 bytes: &[u8],
1479 container_offset: u64,
1480 perms: PermSet,
1481) -> Result<[ByteRange; NUM_PERMS], FileError> {
1482 let ranges = container_section_payload_ranges(bytes, container_offset, perms.len())?;
1483 let mut out = [ByteRange { offset: 0, len: 0 }; NUM_PERMS];
1484 for (perm, range) in perms.iter().zip(ranges) {
1485 out[perm.section_index()] = range;
1486 }
1487 Ok(out)
1488}
1489
1490fn read_uvarint_at<R: RangeReader>(
1491 reader: &R,
1492 absolute_offset: u64,
1493 container_end: u64,
1494) -> Result<(u64, u64), FileError> {
1495 if absolute_offset >= container_end {
1496 return Err(FileError::Container("truncated container varint"));
1497 }
1498 let remaining = container_end - absolute_offset;
1499 let probe_len = remaining.min(10);
1500 let bytes = reader.read_at(absolute_offset, probe_len)?;
1501 read_uvarint(&bytes)
1502 .map(|(value, used)| (value, used as u64))
1503 .ok_or(FileError::Container("truncated container varint"))
1504}
1505
1506fn locate_container_section_ranged<R: RangeReader>(
1509 reader: &R,
1510 container_offset: u64,
1511 container_len: u64,
1512 section_index: usize,
1513 expected_sections: u64,
1514) -> Result<ByteRange, FileError> {
1515 let container_end = checked_end(container_offset, container_len)?;
1516 let (section_count, used) = read_uvarint_at(reader, container_offset, container_end)?;
1517 if section_count != expected_sections {
1518 return Err(FileError::Container("unexpected container section count"));
1519 }
1520 if section_index >= section_count as usize {
1521 return Err(FileError::Container(
1522 "container section index out of bounds",
1523 ));
1524 }
1525
1526 let mut pos = checked_end(container_offset, used)?;
1527 for i in 0..section_count as usize {
1528 let (payload_len, len_used) = read_uvarint_at(reader, pos, container_end)?;
1529 pos = checked_end(pos, len_used)?;
1530 let payload_end = checked_end(pos, payload_len)?;
1531 if payload_end > container_end {
1532 return Err(FileError::Container("section overruns buffer"));
1533 }
1534 if i == section_index {
1535 return Ok(ByteRange {
1536 offset: pos,
1537 len: payload_len,
1538 });
1539 }
1540 pos = payload_end;
1541 }
1542 Err(FileError::Container("container section not found"))
1543}
1544
1545#[allow(clippy::type_complexity)]
1555fn open_index_container_lazy(
1556 reader: &std::sync::Arc<dyn RangeReader + Send + Sync>,
1557 container: ByteRange,
1558 block_codec: u8,
1559 has_synopsis: bool,
1560 read_concurrency: usize,
1561 perms: PermSet,
1562) -> Result<
1563 (
1564 GraphIndex,
1565 [ByteRange; NUM_PERMS],
1566 [Vec<(u32, u32, ByteRange)>; NUM_PERMS],
1567 ),
1568 FileError,
1569> {
1570 let mut index_section_ranges = [ByteRange { offset: 0, len: 0 }; NUM_PERMS];
1571 let mut tile_ranges: [Vec<(u32, u32, ByteRange)>; NUM_PERMS] = Default::default();
1572 #[allow(clippy::type_complexity)]
1573 let mut directories: [Vec<(u32, u32, Option<TileSynopsis>)>; NUM_PERMS] = Default::default();
1574 for (pos, perm) in perms.iter().enumerate() {
1575 let si = perm.section_index();
1576 let section = locate_container_section_ranged(
1577 reader,
1578 container.offset,
1579 container.len,
1580 pos,
1581 perms.len() as u64,
1582 )?;
1583 index_section_ranges[si] = section;
1584 let dir = read_tile_directory_ranged(reader, section)?;
1585 let syn = if has_synopsis {
1588 read_tile_synopsis_ranged(reader, section, &dir)
1589 } else {
1590 vec![None; dir.len()]
1591 };
1592 directories[si] = dir
1593 .iter()
1594 .zip(syn)
1595 .map(|(e, s)| (e.min_a, e.max_a, s))
1596 .collect();
1597 tile_ranges[si] = dir
1598 .into_iter()
1599 .map(|e| {
1600 (
1601 e.min_a,
1602 e.max_a,
1603 ByteRange {
1604 offset: section.offset + e.start,
1605 len: (e.end - e.start),
1606 },
1607 )
1608 })
1609 .collect();
1610 }
1611
1612 let codec = block_codec;
1617 let loader_ranges = tile_ranges.clone();
1618 let loader_reader = reader.clone();
1619 let loader: crate::index::TileLoader = Box::new(move |si, ti| {
1620 let (_, _, range) = loader_ranges.get(si)?.get(ti)?;
1621 let bytes = loader_reader.read_at(range.offset, range.len).ok()?;
1622 decompress(codec, &bytes).ok()
1623 });
1624 let bulk_ranges = tile_ranges.clone();
1625 let bulk_reader = reader.clone();
1626 let bulk: crate::index::TileBulkLoader = Box::new(move |si, tis| {
1627 let section = bulk_ranges.get(si)?;
1628 let want: Option<Vec<ByteRange>> = tis
1629 .iter()
1630 .map(|&ti| section.get(ti).map(|&(_, _, r)| r))
1631 .collect();
1632 let blobs = read_coalesced(bulk_reader.as_ref(), &want?, TILE_COALESCE_GAP)?;
1633 blobs.iter().map(|b| decompress(codec, b).ok()).collect()
1634 });
1635 let mut index =
1636 GraphIndex::from_remote_directories(directories, perms, loader).with_bulk_loader(bulk);
1637 index.set_tile_lens(std::array::from_fn(|si| {
1640 tile_ranges[si]
1641 .iter()
1642 .map(|&(_, _, r)| r.len.min(u32::MAX as u64) as u32)
1643 .collect()
1644 }));
1645 index.set_read_concurrency(read_concurrency);
1649 Ok((index, index_section_ranges, tile_ranges))
1650}
1651
1652pub fn write_file(
1656 dict: &Dictionary,
1657 index: &GraphIndex,
1658 has_quads: bool,
1659 pyramid_meta: &[u8],
1660 pyramid_levels: u16,
1661) -> Vec<u8> {
1662 write_dataset(dict, index, &[], has_quads, pyramid_meta, pyramid_levels)
1663}
1664
1665pub(crate) fn encode_index_container(index: &GraphIndex, codec: u8) -> Vec<u8> {
1671 let payloads: Vec<Vec<u8>> = index
1672 .perms()
1673 .iter()
1674 .map(|perm| encode_tiled_section(index, perm.section_index(), codec))
1675 .collect();
1676 let refs: Vec<&[u8]> = payloads.iter().map(|p| p.as_slice()).collect();
1677 encode_container(&refs, CODEC_NONE)
1678}
1679
1680fn encode_named_graphs(named: &[(String, GraphIndex)], codec: u8) -> Vec<u8> {
1682 let mut out = Vec::new();
1683 write_uvarint(&mut out, named.len() as u64);
1684 for (iri, index) in named {
1685 write_uvarint(&mut out, iri.len() as u64);
1686 out.extend_from_slice(iri.as_bytes());
1687 let container = encode_index_container(index, codec);
1688 write_uvarint(&mut out, container.len() as u64);
1689 out.extend_from_slice(&container);
1690 }
1691 out
1692}
1693
1694fn decode_named_graphs(
1695 bytes: &[u8],
1696 codec: u8,
1697 perms: PermSet,
1698) -> Result<Vec<(String, GraphIndex)>, FileError> {
1699 let (n, mut pos) = read_uvarint(bytes).ok_or(FileError::Container("truncated graph count"))?;
1700 let bound = |start: usize, len: u64| -> Result<usize, FileError> {
1703 start
1704 .checked_add(len as usize)
1705 .filter(|&e| e <= bytes.len())
1706 .ok_or(FileError::Container("named-graph field overruns buffer"))
1707 };
1708 let mut out = Vec::with_capacity((n as usize).min(bytes.len()));
1709 for _ in 0..n {
1710 let (ilen, u1) = read_uvarint(bytes.get(pos..).unwrap_or(&[]))
1711 .ok_or(FileError::Container("truncated iri len"))?;
1712 pos += u1;
1713 let iend = bound(pos, ilen)?;
1714 let iri = String::from_utf8_lossy(&bytes[pos..iend]).into_owned();
1715 pos = iend;
1716 let (clen, u2) = read_uvarint(bytes.get(pos..).unwrap_or(&[]))
1717 .ok_or(FileError::Container("truncated container len"))?;
1718 pos += u2;
1719 let cend = bound(pos, clen)?;
1720 let index = decode_index_container(&bytes[pos..cend], codec, perms)?;
1721 out.push((iri, index));
1722 pos = cend;
1723 }
1724 Ok(out)
1725}
1726
1727pub fn write_dataset(
1730 dict: &Dictionary,
1731 default_index: &GraphIndex,
1732 named: &[(String, GraphIndex)],
1733 has_quads: bool,
1734 pyramid_meta: &[u8],
1735 pyramid_levels: u16,
1736) -> Vec<u8> {
1737 write_dataset_with_metadata(
1738 dict,
1739 default_index,
1740 named,
1741 has_quads,
1742 pyramid_meta,
1743 pyramid_levels,
1744 &[],
1745 &[],
1746 )
1747}
1748
1749pub(crate) fn encode_dict_container(dict: &Dictionary, codec: u8) -> Vec<u8> {
1754 let raw_sections = dict.sections();
1755 let dict_payloads: Vec<Vec<u8>> = raw_sections
1756 .iter()
1757 .map(|raw| encode_chunked_dict_section(raw, codec))
1758 .collect();
1759 encode_container(
1760 &[
1761 dict_payloads[0].as_slice(),
1762 dict_payloads[1].as_slice(),
1763 dict_payloads[2].as_slice(),
1764 dict_payloads[3].as_slice(),
1765 ],
1766 CODEC_NONE,
1767 )
1768}
1769
1770#[allow(clippy::too_many_arguments)]
1781pub fn write_dataset_with_metadata(
1782 dict: &Dictionary,
1783 default_index: &GraphIndex,
1784 named: &[(String, GraphIndex)],
1785 has_quads: bool,
1786 pyramid_meta: &[u8],
1787 pyramid_levels: u16,
1788 metadata: &[u8],
1789 text_index: &[u8],
1790) -> Vec<u8> {
1791 let codec = writer_codec();
1792 let dict_container = encode_dict_container(dict, codec);
1793 write_dataset_from_parts(
1794 &dict_container,
1795 dict.term_count() as u64,
1796 default_index,
1797 named,
1798 has_quads,
1799 dict.has_quoted_triples(),
1800 pyramid_meta,
1801 pyramid_levels,
1802 metadata,
1803 text_index,
1804 codec,
1805 )
1806}
1807
1808#[allow(clippy::too_many_arguments)]
1813pub(crate) fn write_dataset_from_parts(
1814 dict_container: &[u8],
1815 term_count: u64,
1816 default_index: &GraphIndex,
1817 named: &[(String, GraphIndex)],
1818 has_quads: bool,
1819 has_quoted_triples: bool,
1820 pyramid_meta: &[u8],
1821 pyramid_levels: u16,
1822 metadata: &[u8],
1823 text_index: &[u8],
1824 codec: u8,
1825) -> Vec<u8> {
1826 let index_container = encode_index_container(default_index, codec);
1827 let named_section = encode_named_graphs(named, codec);
1828
1829 let meta_section_len = metadata.len() as u64;
1832 let dict_offset = HEADER_LEN as u64 + meta_section_len;
1833 let dict_len = dict_container.len() as u64;
1834 let index_offset = dict_offset + dict_len;
1835 let index_len = index_container.len() as u64;
1836 let pyr_offset = index_offset + index_len;
1837 let pyr_len = pyramid_meta.len() as u64;
1838 let text_offset = pyr_offset + pyr_len;
1840 let text_len = text_index.len() as u64;
1841 let named_offset = text_offset + text_len;
1842 let named_len = if named.is_empty() {
1843 0
1844 } else {
1845 named_section.len() as u64
1846 };
1847
1848 let mut parts: Vec<&[u8]> = Vec::with_capacity(5);
1854 if meta_section_len > 0 {
1855 parts.push(metadata);
1856 }
1857 parts.push(dict_container);
1858 parts.push(&index_container);
1859 parts.push(pyramid_meta);
1860 if text_len > 0 {
1861 parts.push(text_index);
1862 }
1863 if named_len > 0 {
1864 parts.push(&named_section);
1865 }
1866
1867 let schema_meta_len = crate::meta::schema_block_len(pyramid_meta);
1870
1871 let header = Header {
1872 version: crate::header::CURRENT_FORMAT_VERSION,
1873 flags: FLAG_TILE_SYNOPSIS
1874 | if has_quads { FLAG_HAS_QUADS } else { 0 }
1875 | if has_quoted_triples {
1876 FLAG_HAS_QUOTED_TRIPLES
1877 } else {
1878 0
1879 },
1880 metadata_offset: HEADER_LEN as u64,
1881 metadata_len: meta_section_len,
1882 dictionary_offset: dict_offset,
1883 dictionary_len: dict_len,
1884 root_dir_offset: index_offset,
1885 root_dir_len: index_len,
1886 pyramid_meta_offset: if pyr_len > 0 { pyr_offset } else { 0 },
1887 pyramid_meta_len: pyr_len,
1888 dict_codec: codec,
1889 block_codec: codec,
1890 pyramid_levels,
1891 perms: default_index.perms(),
1895 quad_count: default_index.triple_count() as u64
1896 + named
1897 .iter()
1898 .map(|(_, idx)| idx.triple_count() as u64)
1899 .sum::<u64>(),
1900 term_count,
1901 content_hash: content_hash(&parts),
1902 named_graphs_offset: if named_len > 0 { named_offset } else { 0 },
1903 named_graphs_len: named_len,
1904 schema_meta_len,
1905 text_index_offset: if text_len > 0 { text_offset } else { 0 },
1906 text_index_len: text_len,
1907 build_info_offset: 0,
1908 build_info_len: 0,
1909 extra_sections: Vec::new(),
1910 };
1911
1912 let mut out = Vec::with_capacity(
1913 HEADER_LEN
1914 + metadata.len()
1915 + dict_container.len()
1916 + index_container.len()
1917 + pyramid_meta.len()
1918 + text_index.len()
1919 + named_section.len()
1920 + MAGIC.len(),
1921 );
1922 out.extend_from_slice(&header.to_bytes());
1923 if meta_section_len > 0 {
1924 out.extend_from_slice(metadata);
1925 }
1926 out.extend_from_slice(dict_container);
1927 out.extend_from_slice(&index_container);
1928 out.extend_from_slice(pyramid_meta);
1929 if text_len > 0 {
1930 out.extend_from_slice(text_index);
1931 }
1932 if named_len > 0 {
1933 out.extend_from_slice(&named_section);
1934 }
1935 out.extend_from_slice(&MAGIC); out
1937}
1938
1939pub const RDF_TYPE: &str = "<http://www.w3.org/1999/02/22-rdf-syntax-ns#type>";
1941
1942pub fn schema_summary(rete: &Rete) -> Vec<(String, String, String, u32)> {
1949 use std::collections::{BTreeMap, HashMap};
1950 let triples = rete.dump(None);
1951
1952 let mut class_of: HashMap<&str, &str> = HashMap::new();
1953 for (s, p, o) in &triples {
1954 if p == RDF_TYPE {
1955 class_of.insert(s.as_str(), o.as_str());
1956 }
1957 }
1958 let classify = |t: &str| -> String {
1959 if let Some(c) = class_of.get(t) {
1960 (*c).to_string()
1961 } else if t.starts_with('"') {
1962 "(literal)".to_string()
1963 } else {
1964 "(untyped)".to_string()
1965 }
1966 };
1967
1968 let mut counts: BTreeMap<(String, String, String), u32> = BTreeMap::new();
1969 for (s, p, o) in &triples {
1970 if p == RDF_TYPE {
1971 continue; }
1973 *counts
1974 .entry((classify(s), p.clone(), classify(o)))
1975 .or_default() += 1;
1976 }
1977 counts
1978 .into_iter()
1979 .map(|((a, p, b), c)| (a, p, b, c))
1980 .collect()
1981}
1982
1983pub fn schema_classes(rete: &Rete) -> Vec<(String, u32)> {
1987 use std::collections::BTreeMap;
1988 let mut counts: BTreeMap<String, u32> = BTreeMap::new();
1989 for (_s, p, o) in rete.dump(None) {
1990 if p == RDF_TYPE {
1991 *counts.entry(o).or_default() += 1;
1992 }
1993 }
1994 let mut out: Vec<(String, u32)> = counts.into_iter().collect();
1995 out.sort_by(|a, b| b.1.cmp(&a.1).then_with(|| a.0.cmp(&b.0)));
1996 out
1997}
1998
1999pub fn read_metadata_ranged<R: RangeReader>(reader: &R) -> Result<Option<Vec<u8>>, FileError> {
2009 let head = reader.read_at(0, HEADER_LEN as u64)?;
2010 let header = Header::from_bytes(&head)?;
2011 if header.metadata_len == 0 {
2012 return Ok(None);
2013 }
2014 let bytes = reader.read_at(header.metadata_offset, header.metadata_len)?;
2015 Ok(Some(bytes))
2016}
2017
2018#[allow(clippy::type_complexity)]
2028pub fn read_card_and_build_info_ranged<R: RangeReader>(
2029 reader: &R,
2030) -> Result<(Option<Vec<u8>>, Option<Vec<u8>>), FileError> {
2031 read_card_and_build_info_with_header(reader).map(|(_, m, b)| (m, b))
2032}
2033
2034#[allow(clippy::type_complexity)]
2045pub fn read_card_and_build_info_with_header<R: RangeReader>(
2046 reader: &R,
2047) -> Result<(Header, Option<Vec<u8>>, Option<Vec<u8>>), FileError> {
2048 let head = reader.read_at(0, HEADER_LEN as u64)?;
2049 let header = Header::from_bytes(&head)?;
2050 let meta = (header.metadata_offset, header.metadata_len);
2051 let build = (header.build_info_offset, header.build_info_len);
2052 if meta.1 > 0 && build.1 > 0 && build.0 == meta.0 + meta.1 {
2053 let both = reader.read_at(meta.0, meta.1 + build.1)?;
2055 let (m, b) = both.split_at(meta.1 as usize);
2056 return Ok((header, Some(m.to_vec()), Some(b.to_vec())));
2057 }
2058 let fetch = |off: u64, len: u64| -> Result<Option<Vec<u8>>, FileError> {
2059 if len == 0 {
2060 return Ok(None);
2061 }
2062 Ok(Some(reader.read_at(off, len)?))
2063 };
2064 Ok((header, fetch(meta.0, meta.1)?, fetch(build.0, build.1)?))
2065}
2066
2067pub fn read_build_info(bytes: &[u8]) -> Result<Option<Vec<u8>>, FileError> {
2072 let header = Header::from_bytes(bytes)?;
2073 if header.build_info_len == 0 {
2074 return Ok(None);
2075 }
2076 let start = header.build_info_offset as usize;
2077 let end = start
2078 .checked_add(header.build_info_len as usize)
2079 .filter(|&e| e <= bytes.len())
2080 .ok_or(FileError::Container("build-info section overruns buffer"))?;
2081 Ok(Some(bytes[start..end].to_vec()))
2082}
2083
2084#[derive(Debug, Clone, PartialEq, Eq)]
2091pub struct BuildInfoPlan {
2092 pub header: [u8; HEADER_LEN],
2094 pub insert: u64,
2097 pub tail_start: u64,
2101 pub new_len: u64,
2103}
2104
2105pub fn plan_build_info(
2127 head: &[u8],
2128 file_len: u64,
2129 info_len: u64,
2130) -> Result<BuildInfoPlan, FileError> {
2131 let mut header = Header::from_bytes(head)?;
2132 let insert = HEADER_LEN as u64 + header.metadata_len;
2135 let old_len = header.build_info_len;
2136 if old_len > 0 && header.build_info_offset != insert {
2137 return Err(FileError::Container(
2138 "existing build-info section is not adjacent to the metadata",
2139 ));
2140 }
2141 let tail_start = insert.saturating_add(old_len);
2142 if tail_start > file_len || insert > file_len {
2143 return Err(FileError::Container("build-info splice out of bounds"));
2144 }
2145
2146 let shift = |off: &mut u64, len: u64| {
2148 if len > 0 && *off >= tail_start {
2149 *off = *off - old_len + info_len;
2150 }
2151 };
2152 shift(&mut header.dictionary_offset, header.dictionary_len);
2153 shift(&mut header.root_dir_offset, header.root_dir_len);
2154 shift(&mut header.pyramid_meta_offset, header.pyramid_meta_len);
2155 shift(&mut header.named_graphs_offset, header.named_graphs_len);
2156 shift(&mut header.text_index_offset, header.text_index_len);
2157 for s in &mut header.extra_sections {
2158 if s.length > 0 && s.offset >= tail_start {
2159 s.offset = s.offset - old_len + info_len;
2160 }
2161 }
2162 if info_len == 0 {
2163 header.build_info_offset = 0;
2164 header.build_info_len = 0;
2165 } else {
2166 header.build_info_offset = insert;
2167 header.build_info_len = info_len;
2168 }
2169
2170 Ok(BuildInfoPlan {
2171 header: header.to_bytes(),
2172 insert,
2173 tail_start,
2174 new_len: file_len - old_len + info_len,
2175 })
2176}
2177
2178pub fn attach_build_info(image: &[u8], info: &[u8]) -> Result<Vec<u8>, FileError> {
2184 let plan = plan_build_info(image, image.len() as u64, info.len() as u64)?;
2185 let mut out = Vec::with_capacity(plan.new_len as usize);
2186 out.extend_from_slice(&plan.header);
2187 out.extend_from_slice(&image[HEADER_LEN..plan.insert as usize]);
2188 out.extend_from_slice(info);
2189 out.extend_from_slice(&image[plan.tail_start as usize..]);
2190 debug_assert_eq!(out.len() as u64, plan.new_len);
2191 Ok(out)
2192}
2193
2194pub fn replace_metadata(image: &[u8], metadata: &[u8]) -> Result<Vec<u8>, FileError> {
2214 let mut header = Header::from_bytes(image)?;
2215 let insert = HEADER_LEN as u64;
2218 if header.metadata_len > 0 && header.metadata_offset != insert {
2219 return Err(FileError::Container(
2220 "metadata section does not sit immediately after the header",
2221 ));
2222 }
2223 let old_len = header.metadata_len;
2224 let tail_start = (insert + old_len) as usize;
2225 if tail_start > image.len() {
2226 return Err(FileError::Container("metadata splice out of bounds"));
2227 }
2228 let new_len = metadata.len() as u64;
2229
2230 let shift = |off: &mut u64, len: u64| {
2231 if len > 0 && *off >= insert + old_len {
2232 *off = *off - old_len + new_len;
2233 }
2234 };
2235 shift(&mut header.dictionary_offset, header.dictionary_len);
2236 shift(&mut header.root_dir_offset, header.root_dir_len);
2237 shift(&mut header.pyramid_meta_offset, header.pyramid_meta_len);
2238 shift(&mut header.named_graphs_offset, header.named_graphs_len);
2239 shift(&mut header.text_index_offset, header.text_index_len);
2240 shift(&mut header.build_info_offset, header.build_info_len);
2241 for s in &mut header.extra_sections {
2242 if s.length > 0 && s.offset >= insert + old_len {
2243 s.offset = s.offset - old_len + new_len;
2244 }
2245 }
2246 header.metadata_offset = insert;
2247 header.metadata_len = new_len;
2248
2249 let mut out = Vec::with_capacity(image.len() - old_len as usize + metadata.len());
2250 out.extend_from_slice(&header.to_bytes());
2251 out.extend_from_slice(metadata);
2252 out.extend_from_slice(&image[tail_start..]);
2253 header.content_hash = hash_of_sections(&header, &out)?;
2257 out[..HEADER_LEN].copy_from_slice(&header.to_bytes());
2258 Ok(out)
2259}
2260
2261fn hash_of_sections(header: &Header, bytes: &[u8]) -> Result<[u8; 16], FileError> {
2267 let slice = |off: u64, len: u64| -> Result<&[u8], FileError> {
2268 let end = off.saturating_add(len);
2275 bytes
2276 .get(off as usize..end as usize)
2277 .ok_or(FileError::Container("section overruns buffer"))
2278 };
2279 let d = slice(header.dictionary_offset, header.dictionary_len)?;
2280 let i = slice(header.root_dir_offset, header.root_dir_len)?;
2281 let m = if header.pyramid_meta_len > 0 {
2282 slice(header.pyramid_meta_offset, header.pyramid_meta_len)?
2283 } else {
2284 &[]
2285 };
2286 let mut parts: Vec<&[u8]> = Vec::with_capacity(6);
2287 if header.metadata_len > 0 {
2288 parts.push(slice(header.metadata_offset, header.metadata_len)?);
2289 }
2290 parts.push(d);
2291 parts.push(i);
2292 parts.push(m);
2293 if header.text_index_len > 0 {
2294 parts.push(slice(header.text_index_offset, header.text_index_len)?);
2295 }
2296 if header.named_graphs_len > 0 {
2297 parts.push(slice(header.named_graphs_offset, header.named_graphs_len)?);
2298 }
2299 Ok(content_hash(&parts))
2300}
2301
2302pub fn verify(bytes: &[u8]) -> Result<bool, FileError> {
2305 let header = Header::from_bytes(bytes)?;
2306 Ok(hash_of_sections(&header, bytes)? == header.content_hash)
2307}
2308
2309pub fn verify_ranged(reader: &dyn crate::reader::RangeReader) -> Result<bool, FileError> {
2315 const SLICE: u64 = 8 << 20;
2318 let len = reader.len();
2319 if len < HEADER_LEN as u64 {
2320 return Err(FileError::Container("file shorter than its header"));
2321 }
2322 let header = Header::from_bytes(&reader.read_at(0, HEADER_LEN as u64)?)?;
2323 let mut parts: Vec<(u64, u64)> = Vec::with_capacity(6);
2324 if header.metadata_len > 0 {
2325 parts.push((header.metadata_offset, header.metadata_len));
2326 }
2327 parts.push((header.dictionary_offset, header.dictionary_len));
2328 parts.push((header.root_dir_offset, header.root_dir_len));
2329 if header.pyramid_meta_len > 0 {
2330 parts.push((header.pyramid_meta_offset, header.pyramid_meta_len));
2331 }
2332 if header.text_index_len > 0 {
2333 parts.push((header.text_index_offset, header.text_index_len));
2334 }
2335 if header.named_graphs_len > 0 {
2336 parts.push((header.named_graphs_offset, header.named_graphs_len));
2337 }
2338 let mut h = blake3::Hasher::new();
2339 for (off, n) in parts {
2340 if off.checked_add(n).is_none_or(|end| end > len) {
2343 return Err(FileError::Container("section overruns buffer"));
2344 }
2345 let mut at = off;
2346 while at < off + n {
2347 let take = (off + n - at).min(SLICE);
2348 h.update(&reader.read_at(at, take)?);
2349 at += take;
2350 }
2351 }
2352 let mut out = [0u8; 16];
2353 out.copy_from_slice(&h.finalize().as_bytes()[..16]);
2354 Ok(out == header.content_hash)
2355}
2356
2357type PyramidLoader = Box<dyn Fn() -> Option<PyramidMeta> + Send + Sync>;
2359
2360enum PyramidSlot {
2366 Resident(Option<PyramidMeta>),
2367 Lazy {
2368 loader: PyramidLoader,
2369 cell: std::sync::OnceLock<Option<PyramidMeta>>,
2370 },
2371}
2372
2373type TextIndexLoader = Box<dyn Fn() -> Option<crate::text_index::TextIndex> + Send + Sync>;
2376
2377type TokenTableProbe = Box<dyn Fn() -> Option<u64> + Send + Sync>;
2380
2381enum TextIndexSlot {
2391 Resident {
2392 index: Option<crate::text_index::TextIndex>,
2393 token_table_len: Option<u64>,
2396 },
2397 Lazy {
2398 loader: TextIndexLoader,
2399 cell: std::sync::OnceLock<Option<crate::text_index::TextIndex>>,
2400 token_table: TokenTableProbe,
2402 token_table_cell: std::sync::OnceLock<Option<u64>>,
2403 },
2404}
2405
2406impl TextIndexSlot {
2407 fn index(&self) -> Option<&crate::text_index::TextIndex> {
2409 match self {
2410 TextIndexSlot::Resident { index, .. } => index.as_ref(),
2411 TextIndexSlot::Lazy { loader, cell, .. } => cell.get_or_init(loader).as_ref(),
2412 }
2413 }
2414
2415 fn token_table_len(&self) -> Option<u64> {
2418 match self {
2419 TextIndexSlot::Resident {
2420 token_table_len, ..
2421 } => *token_table_len,
2422 TextIndexSlot::Lazy {
2423 token_table,
2424 token_table_cell,
2425 ..
2426 } => *token_table_cell.get_or_init(token_table),
2427 }
2428 }
2429}
2430
2431fn resident_text_index_slot(section: Option<&[u8]>, codec: u8) -> Result<TextIndexSlot, FileError> {
2434 let Some(section) = section else {
2435 return Ok(TextIndexSlot::Resident {
2436 index: None,
2437 token_table_len: None,
2438 });
2439 };
2440 Ok(TextIndexSlot::Resident {
2441 index: Some(
2442 crate::text_index::TextIndex::from_section(section, codec)
2443 .map_err(|_| FileError::Container("malformed text index"))?,
2444 ),
2445 token_table_len: crate::text_index::TextIndex::postings_base(section).map(|b| b as u64),
2446 })
2447}
2448
2449pub fn read_text_index_token_table_len_ranged<R: RangeReader + ?Sized>(
2462 reader: &R,
2463 header: &Header,
2464) -> Option<u64> {
2465 if header.text_index_len == 0 {
2466 return None;
2467 }
2468 read_token_table_len(reader, header.text_index_offset, header.text_index_len)
2469}
2470
2471fn read_token_table_len<R: RangeReader + ?Sized>(reader: &R, off: u64, len: u64) -> Option<u64> {
2475 let head = reader.read_at(off, 10u64.min(len)).ok()?;
2476 let (ttlen, n) = crate::varint::read_uvarint(&head)?;
2477 Some((n as u64 + ttlen).min(len))
2478}
2479
2480const NAMED_GRAPH_RESIDENT_MAX: u64 = 1 << 20; const NAMED_SLAB: usize = 1024;
2491
2492const NAMED_WALK_CHUNK: u64 = 64 * 1024;
2496
2497const NAMED_WALK_CHUNK_MAX: u64 = 8 * 1024 * 1024;
2510
2511#[derive(Default)]
2513struct NamedEntry {
2514 meta: std::sync::OnceLock<(String, ByteRange)>,
2517 index: std::sync::OnceLock<Box<GraphIndex>>,
2520}
2521
2522#[derive(Default)]
2526struct NamedWalk {
2527 next: usize,
2529 pos: u64,
2533 buf: Vec<u8>,
2540 buf_off: u64,
2541 chunk: u64,
2548 big_seen: bool,
2555}
2556
2557struct LazyNamedGraphs {
2571 reader: std::sync::Arc<dyn RangeReader + Send + Sync>,
2572 section: ByteRange,
2574 codec: u8,
2575 has_synopsis: bool,
2576 read_concurrency: usize,
2577 perms: PermSet,
2580 #[allow(clippy::type_complexity)]
2584 dir: std::sync::OnceLock<(usize, Box<[std::sync::OnceLock<Box<[NamedEntry]>>]>)>,
2585 walk: std::sync::Mutex<NamedWalk>,
2586 failed: std::sync::atomic::AtomicBool,
2587 tile_cap: std::sync::atomic::AtomicU64,
2593}
2594
2595impl LazyNamedGraphs {
2596 fn new(
2597 reader: std::sync::Arc<dyn RangeReader + Send + Sync>,
2598 section: ByteRange,
2599 codec: u8,
2600 has_synopsis: bool,
2601 read_concurrency: usize,
2602 perms: PermSet,
2603 ) -> Self {
2604 LazyNamedGraphs {
2605 reader,
2606 section,
2607 codec,
2608 has_synopsis,
2609 read_concurrency,
2610 perms,
2611 dir: std::sync::OnceLock::new(),
2612 walk: std::sync::Mutex::new(NamedWalk::default()),
2613 failed: std::sync::atomic::AtomicBool::new(false),
2614 tile_cap: std::sync::atomic::AtomicU64::new(u64::MAX),
2615 }
2616 }
2617
2618 fn set_tile_cap(&self, cap: u64) {
2622 self.tile_cap
2623 .store(cap, std::sync::atomic::Ordering::Relaxed);
2624 self.for_each_opened(|g| {
2625 if g.is_lazy() {
2626 g.set_cache_cap(cap);
2627 }
2628 });
2629 }
2630
2631 fn fail(&self) {
2633 self.failed
2634 .store(true, std::sync::atomic::Ordering::Relaxed);
2635 }
2636
2637 #[allow(clippy::type_complexity)]
2639 fn directory(&self) -> Option<&(usize, Box<[std::sync::OnceLock<Box<[NamedEntry]>>]>)> {
2640 if let Some(d) = self.dir.get() {
2641 return Some(d);
2642 }
2643 let end = self.section.offset.checked_add(self.section.len)?;
2644 let (n, used) = match read_uvarint_at(&self.reader, self.section.offset, end) {
2645 Ok(v) => v,
2646 Err(_) => {
2647 self.fail();
2648 return None;
2649 }
2650 };
2651 if n > self.section.len / 2 {
2654 self.fail();
2655 return None;
2656 }
2657 let n = n as usize;
2658 let slabs = n.div_ceil(NAMED_SLAB);
2659 let table: Box<[std::sync::OnceLock<Box<[NamedEntry]>>]> =
2660 (0..slabs).map(|_| std::sync::OnceLock::new()).collect();
2661 let _ = self.dir.set((n, table));
2662 {
2663 let mut w = self.walk.lock().unwrap();
2666 if w.pos == 0 {
2667 w.pos = self.section.offset + used;
2668 }
2669 }
2670 self.dir.get()
2671 }
2672
2673 fn count(&self) -> usize {
2674 self.directory().map(|(n, _)| *n).unwrap_or(0)
2675 }
2676
2677 fn entry(&self, i: usize) -> Option<&NamedEntry> {
2679 let (n, table) = self.directory()?;
2680 if i >= *n {
2681 return None;
2682 }
2683 let slab = table[i / NAMED_SLAB].get_or_init(|| {
2684 let len = NAMED_SLAB.min(n - (i / NAMED_SLAB) * NAMED_SLAB);
2685 (0..len).map(|_| NamedEntry::default()).collect()
2686 });
2687 slab.get(i % NAMED_SLAB)
2688 }
2689
2690 fn next_read_len(w: &mut NamedWalk, pos: u64, end: u64, need: u64) -> u64 {
2698 let cur = w.chunk.max(NAMED_WALK_CHUNK);
2699 w.chunk = (cur * 2).min(NAMED_WALK_CHUNK_MAX);
2700 cur.max(need).min(end - pos)
2701 }
2702
2703 fn ensure_meta(&self, upto: usize, exhaustive: bool) -> Option<()> {
2718 let target = self.entry(upto)?;
2719 if target.meta.get().is_some() {
2720 return Some(());
2721 }
2722 let end = self.section.offset.checked_add(self.section.len)?;
2723 let mut w = self.walk.lock().unwrap();
2724 if exhaustive && !w.big_seen {
2725 w.chunk = NAMED_WALK_CHUNK_MAX;
2726 }
2727 while w.next <= upto {
2728 let pos = w.pos;
2729 if pos >= end {
2730 self.fail();
2732 return None;
2733 }
2734 let have =
2737 |b: &[u8], off: u64, need: u64| pos >= off && pos + need <= off + b.len() as u64;
2738 if !have(&w.buf, w.buf_off, 20.min(end - pos)) {
2739 let len = Self::next_read_len(&mut w, pos, end, 20.min(end - pos));
2740 w.buf = match self.reader.read_at(pos, len) {
2741 Ok(b) => b,
2742 Err(_) => {
2743 self.fail();
2744 return None;
2745 }
2746 };
2747 w.buf_off = pos;
2748 }
2749 let rel = (pos - w.buf_off) as usize;
2750 let Some((ilen, u1)) = read_uvarint(&w.buf[rel..]) else {
2751 self.fail();
2752 return None;
2753 };
2754 let header_need = u1 as u64 + ilen + 10; if pos + u1 as u64 + ilen > end {
2756 self.fail(); return None;
2758 }
2759 if !have(&w.buf, w.buf_off, header_need.min(end - pos)) {
2760 let len = Self::next_read_len(&mut w, pos, end, header_need);
2761 w.buf = match self.reader.read_at(pos, len) {
2762 Ok(b) => b,
2763 Err(_) => {
2764 self.fail();
2765 return None;
2766 }
2767 };
2768 w.buf_off = pos;
2769 }
2770 let rel = (pos - w.buf_off) as usize;
2771 let istart = rel + u1;
2772 let iend = istart + ilen as usize;
2773 let iri = String::from_utf8_lossy(&w.buf[istart..iend]).into_owned();
2774 let Some((clen, u2)) = read_uvarint(&w.buf[iend..]) else {
2775 self.fail();
2776 return None;
2777 };
2778 let cstart = pos + u1 as u64 + ilen + u2 as u64;
2779 let cend = match cstart.checked_add(clen) {
2780 Some(e) if e <= end => e,
2781 _ => {
2782 self.fail(); return None;
2784 }
2785 };
2786 if clen > NAMED_GRAPH_RESIDENT_MAX {
2787 w.big_seen = true;
2792 w.chunk = NAMED_WALK_CHUNK;
2793 }
2794 let range = ByteRange {
2795 offset: cstart,
2796 len: clen,
2797 };
2798 if let Some(e) = self.entry(w.next) {
2799 let _ = e.meta.set((iri, range));
2800 }
2801 w.pos = cend;
2802 w.next += 1;
2803 }
2804 Some(())
2805 }
2806
2807 fn name_at(&self, i: usize, exhaustive: bool) -> Option<&str> {
2810 self.ensure_meta(i, exhaustive)?;
2811 self.entry(i)?.meta.get().map(|(iri, _)| iri.as_str())
2812 }
2813
2814 fn graph_at(&self, i: usize, exhaustive: bool) -> Option<(&str, &GraphIndex)> {
2817 self.ensure_meta(i, exhaustive)?;
2818 let e = self.entry(i)?;
2819 let (iri, range) = e.meta.get()?;
2820 if e.index.get().is_none() {
2821 let opened = self.open_graph(*range)?;
2822 let _ = e.index.set(Box::new(opened));
2823 }
2824 Some((iri.as_str(), e.index.get()?.as_ref()))
2825 }
2826
2827 fn container_bytes(&self, range: ByteRange) -> Option<Vec<u8>> {
2831 {
2832 let w = self.walk.lock().unwrap();
2833 let buf_end = w.buf_off + w.buf.len() as u64;
2834 if range.offset >= w.buf_off && range.offset + range.len <= buf_end {
2835 let a = (range.offset - w.buf_off) as usize;
2836 return Some(w.buf[a..a + range.len as usize].to_vec());
2837 }
2838 }
2839 match self.reader.read_at(range.offset, range.len) {
2840 Ok(b) => Some(b),
2841 Err(_) => {
2842 self.fail();
2843 None
2844 }
2845 }
2846 }
2847
2848 fn open_graph(&self, range: ByteRange) -> Option<GraphIndex> {
2853 if range.len <= NAMED_GRAPH_RESIDENT_MAX {
2854 let bytes = self.container_bytes(range)?;
2855 match decode_index_container(&bytes, self.codec, self.perms) {
2856 Ok(g) => Some(g),
2857 Err(_) => {
2858 self.fail();
2859 None
2860 }
2861 }
2862 } else {
2863 match open_index_container_lazy(
2864 &self.reader,
2865 range,
2866 self.codec,
2867 self.has_synopsis,
2868 self.read_concurrency,
2869 self.perms,
2870 ) {
2871 Ok((g, _, _)) => {
2872 let cap = self.tile_cap.load(std::sync::atomic::Ordering::Relaxed);
2876 if cap != u64::MAX {
2877 g.set_cache_cap(cap);
2878 }
2879 Some(g)
2880 }
2881 Err(_) => {
2882 self.fail();
2883 None
2884 }
2885 }
2886 }
2887 }
2888
2889 fn release(&mut self, iri: &str) {
2895 let Some((_, table)) = self.dir.get_mut() else {
2896 return;
2897 };
2898 for slab in table.iter_mut() {
2899 let Some(entries) = slab.get_mut() else {
2900 continue;
2901 };
2902 for e in entries.iter_mut() {
2903 let matches = e
2904 .meta
2905 .get()
2906 .map(|(name, _)| name.as_str() == iri)
2907 .unwrap_or(false);
2908 if matches {
2909 e.index.take();
2910 return;
2911 }
2912 }
2913 }
2914 }
2915
2916 fn for_each_opened(&self, mut f: impl FnMut(&GraphIndex)) {
2918 if let Some((_, table)) = self.dir.get() {
2919 for slab in table.iter().filter_map(|s| s.get()) {
2920 for e in slab.iter() {
2921 if let Some(g) = e.index.get() {
2922 f(g);
2923 }
2924 }
2925 }
2926 }
2927 }
2928}
2929
2930enum NamedGraphsSlot {
2935 Resident(Vec<(String, GraphIndex)>),
2936 Lazy(LazyNamedGraphs),
2937}
2938
2939impl NamedGraphsSlot {
2940 fn directory_bytes(&self) -> u64 {
2944 match self {
2945 NamedGraphsSlot::Resident(v) => v
2946 .iter()
2947 .map(|(iri, g)| iri.capacity() as u64 + g.directory_bytes())
2948 .sum(),
2949 NamedGraphsSlot::Lazy(l) => {
2950 let mut bytes = l.walk.lock().map(|w| w.buf.capacity() as u64).unwrap_or(0);
2951 if let Some((_, table)) = l.dir.get() {
2952 bytes += table.len() as u64
2953 * std::mem::size_of::<std::sync::OnceLock<Box<[NamedEntry]>>>() as u64;
2954 for slab in table.iter().filter_map(|s| s.get()) {
2955 bytes += slab.len() as u64 * std::mem::size_of::<NamedEntry>() as u64;
2956 for e in slab.iter() {
2957 if let Some((iri, _)) = e.meta.get() {
2958 bytes += iri.capacity() as u64;
2959 }
2960 if let Some(g) = e.index.get() {
2961 bytes += g.directory_bytes();
2962 }
2963 }
2964 }
2965 }
2966 bytes
2967 }
2968 }
2969 }
2970
2971 fn count(&self) -> usize {
2972 match self {
2973 NamedGraphsSlot::Resident(v) => v.len(),
2974 NamedGraphsSlot::Lazy(l) => l.count(),
2975 }
2976 }
2977
2978 fn name_at(&self, i: usize, exhaustive: bool) -> Option<&str> {
2981 match self {
2982 NamedGraphsSlot::Resident(v) => v.get(i).map(|(iri, _)| iri.as_str()),
2983 NamedGraphsSlot::Lazy(l) => l.name_at(i, exhaustive),
2984 }
2985 }
2986
2987 fn graph_at(&self, i: usize, exhaustive: bool) -> Option<(&str, &GraphIndex)> {
2988 match self {
2989 NamedGraphsSlot::Resident(v) => v.get(i).map(|(iri, g)| (iri.as_str(), g)),
2990 NamedGraphsSlot::Lazy(l) => l.graph_at(i, exhaustive),
2991 }
2992 }
2993
2994 fn find(&self, iri: &str) -> Option<&GraphIndex> {
2995 match self {
2996 NamedGraphsSlot::Resident(v) => v.iter().find(|(name, _)| name == iri).map(|(_, g)| g),
2997 NamedGraphsSlot::Lazy(l) => {
2998 for i in 0..l.count() {
3003 if l.name_at(i, false)? == iri {
3004 return l.graph_at(i, false).map(|(_, g)| g);
3005 }
3006 }
3007 None
3008 }
3009 }
3010 }
3011
3012 fn load_incomplete(&self) -> bool {
3013 match self {
3014 NamedGraphsSlot::Resident(v) => v.iter().any(|(_, g)| g.load_incomplete()),
3015 NamedGraphsSlot::Lazy(l) => {
3016 if l.failed.load(std::sync::atomic::Ordering::Relaxed) {
3017 return true;
3018 }
3019 let mut bad = false;
3020 l.for_each_opened(|g| bad |= g.load_incomplete());
3021 bad
3022 }
3023 }
3024 }
3025
3026 fn reset_load_failures(&self) {
3027 match self {
3028 NamedGraphsSlot::Resident(v) => {
3029 for (_, g) in v {
3030 g.reset_load_failure();
3031 }
3032 }
3033 NamedGraphsSlot::Lazy(l) => {
3034 l.failed.store(false, std::sync::atomic::Ordering::Relaxed);
3035 l.for_each_opened(|g| g.reset_load_failure());
3036 }
3037 }
3038 }
3039}
3040
3041pub struct Rete {
3043 header: Header,
3044 dict: Dictionary,
3045 index: GraphIndex,
3046 index_section_ranges: [ByteRange; NUM_PERMS],
3047 tile_ranges: [Vec<(u32, u32, ByteRange)>; NUM_PERMS],
3051 pyramid: PyramidSlot,
3052 text_index: TextIndexSlot,
3053 named_graphs: NamedGraphsSlot,
3054 metadata: Vec<u8>,
3059 service_client: Option<Box<dyn crate::service::ServiceClient>>,
3064 service_error: std::sync::Mutex<Option<String>>,
3068 spill_threshold: std::sync::atomic::AtomicU64,
3071}
3072
3073impl Rete {
3074 pub fn open(bytes: &[u8]) -> Result<Self, FileError> {
3077 let header = Header::from_bytes(bytes)?;
3078
3079 let region = |off: u64, len: u64| -> Result<&[u8], FileError> {
3083 let start = off as usize;
3084 let end = start
3085 .checked_add(len as usize)
3086 .filter(|&e| e <= bytes.len())
3087 .ok_or(FileError::Container("section range out of bounds"))?;
3088 Ok(&bytes[start..end])
3089 };
3090
3091 let dict = decode_dictionary_container(
3092 region(header.dictionary_offset, header.dictionary_len)?,
3093 header.dict_codec,
3094 )?;
3095
3096 let index_bytes = region(header.root_dir_offset, header.root_dir_len)?;
3097 let index = decode_index_container(index_bytes, header.block_codec, header.perms)?;
3098 let index_section_ranges =
3099 decode_index_section_ranges(index_bytes, header.root_dir_offset, header.perms)?;
3100
3101 let pyramid = PyramidSlot::Resident(if header.pyramid_meta_len > 0 {
3102 Some(
3103 PyramidMeta::decode(region(header.pyramid_meta_offset, header.pyramid_meta_len)?)
3104 .map_err(|_| FileError::Container("malformed pyramid meta"))?,
3105 )
3106 } else {
3107 None
3108 });
3109
3110 let text_index = resident_text_index_slot(
3113 if header.text_index_len > 0 {
3114 Some(region(header.text_index_offset, header.text_index_len)?)
3115 } else {
3116 None
3117 },
3118 header.block_codec,
3119 )?;
3120
3121 let named_graphs = NamedGraphsSlot::Resident(if header.named_graphs_len > 0 {
3122 decode_named_graphs(
3123 region(header.named_graphs_offset, header.named_graphs_len)?,
3124 header.block_codec,
3125 header.perms,
3126 )?
3127 } else {
3128 Vec::new()
3129 });
3130
3131 let metadata = if header.metadata_len > 0 {
3132 region(header.metadata_offset, header.metadata_len)?.to_vec()
3133 } else {
3134 Vec::new()
3135 };
3136
3137 let tile_ranges =
3138 tile_file_ranges(index_bytes, header.root_dir_offset, &index_section_ranges);
3139 Ok(Self {
3140 header,
3141 dict,
3142 index,
3143 index_section_ranges,
3144 tile_ranges,
3145 pyramid,
3146 text_index,
3147 named_graphs,
3148 metadata,
3149 service_client: None,
3150 service_error: std::sync::Mutex::new(None),
3151 spill_threshold: std::sync::atomic::AtomicU64::new(u64::MAX),
3152 })
3153 }
3154
3155 pub fn set_service_client(&mut self, client: Box<dyn crate::service::ServiceClient>) {
3160 self.service_client = Some(client);
3161 }
3162
3163 pub(crate) fn service_client(&self) -> Option<&dyn crate::service::ServiceClient> {
3164 self.service_client.as_deref()
3165 }
3166
3167 pub(crate) fn record_service_error(&self, msg: &str) {
3169 let mut e = self.service_error.lock().unwrap();
3170 if e.is_none() {
3171 *e = Some(msg.to_string());
3172 }
3173 }
3174
3175 pub(crate) fn take_service_error(&self) -> Option<String> {
3178 self.service_error.lock().unwrap().take()
3179 }
3180
3181 pub fn header(&self) -> &Header {
3182 &self.header
3183 }
3184
3185 pub fn file_layout(&self) -> Vec<LayoutSegment> {
3191 let h = &self.header;
3192 let seg = |kind: &'static str, label: String, offset: u64, len: u64| LayoutSegment {
3193 kind,
3194 label,
3195 offset,
3196 len,
3197 };
3198 let mut out = vec![seg(
3204 "header",
3205 format!("header (fixed {} bytes)", crate::header::HEADER_LEN),
3206 0,
3207 crate::header::HEADER_LEN as u64,
3208 )];
3209 if h.metadata_len > 0 {
3210 out.push(seg(
3211 "metadata",
3212 "metadata (dataset card)".into(),
3213 h.metadata_offset,
3214 h.metadata_len,
3215 ));
3216 }
3217 out.push(seg(
3218 "dictionary",
3219 "dictionary (4 front-coded term sections)".into(),
3220 h.dictionary_offset,
3221 h.dictionary_len,
3222 ));
3223 for (si, perm) in crate::index::ALL_PERMS.into_iter().enumerate() {
3224 let sec = self.index_section_ranges[si];
3225 if sec.len == 0 {
3226 continue;
3227 }
3228 let first_tile = self.tile_ranges[si]
3229 .first()
3230 .map(|&(_, _, r)| r.offset)
3231 .unwrap_or(sec.offset + sec.len);
3232 if first_tile > sec.offset {
3233 out.push(seg(
3234 "directory",
3235 format!("{} tile directory", perm.name()),
3236 sec.offset,
3237 first_tile - sec.offset,
3238 ));
3239 }
3240 for (ti, &(min_a, max_a, r)) in self.tile_ranges[si].iter().enumerate() {
3241 out.push(seg(
3242 "tile",
3243 format!("{} tile {ti} (leading ids {min_a}..{max_a})", perm.name()),
3244 r.offset,
3245 r.len,
3246 ));
3247 }
3248 }
3249 if h.pyramid_meta_len > 0 {
3250 out.push(seg(
3251 "pyramid",
3252 "pyramid summary (communities + superedges)".into(),
3253 h.pyramid_meta_offset,
3254 h.pyramid_meta_len,
3255 ));
3256 }
3257 if h.named_graphs_len > 0 {
3258 out.push(seg(
3259 "named-graphs",
3260 format!("named graphs ({})", self.named_graphs.count()),
3261 h.named_graphs_offset,
3262 h.named_graphs_len,
3263 ));
3264 }
3265 out.sort_by_key(|s| s.offset);
3266 out
3267 }
3268
3269 pub fn metadata(&self) -> Option<&[u8]> {
3274 if self.metadata.is_empty() {
3275 None
3276 } else {
3277 Some(&self.metadata)
3278 }
3279 }
3280
3281 pub fn dictionary(&self) -> &Dictionary {
3282 &self.dict
3283 }
3284
3285 pub fn pyramid(&self) -> Option<&PyramidMeta> {
3287 match &self.pyramid {
3288 PyramidSlot::Resident(p) => p.as_ref(),
3289 PyramidSlot::Lazy { loader, cell } => cell.get_or_init(loader).as_ref(),
3291 }
3292 }
3293
3294 pub fn pyramid_if_loaded(&self) -> Option<&PyramidMeta> {
3299 match &self.pyramid {
3300 PyramidSlot::Resident(p) => p.as_ref(),
3301 PyramidSlot::Lazy { cell, .. } => cell.get().and_then(|o| o.as_ref()),
3302 }
3303 }
3304
3305 pub fn predicate_stats(&self) -> &[crate::meta::PredStat] {
3309 self.pyramid_if_loaded()
3310 .map(|p| p.predicate_stats.as_slice())
3311 .unwrap_or(&[])
3312 }
3313
3314 pub fn char_sets(&self) -> &[crate::meta::CharSet] {
3317 self.pyramid_if_loaded()
3318 .map(|p| p.char_sets.as_slice())
3319 .unwrap_or(&[])
3320 }
3321
3322 pub fn label_index(&self) -> &[crate::meta::LabelEntry] {
3325 self.pyramid_if_loaded()
3326 .map(|p| p.label_index.as_slice())
3327 .unwrap_or(&[])
3328 }
3329
3330 pub fn prefix_search(&self, prefix: &str, limit: usize) -> Vec<(String, String)> {
3336 let Some(pyr) = self.pyramid() else {
3337 return Vec::new();
3338 };
3339 pyr.prefix_search(prefix, limit)
3340 .into_iter()
3341 .filter_map(|e| {
3342 self.dict
3343 .subject_term(e.subject)
3344 .map(|iri| (e.label.clone(), iri))
3345 })
3346 .collect()
3347 }
3348
3349 pub(crate) fn text_index(&self) -> Option<&crate::text_index::TextIndex> {
3352 self.text_index.index()
3353 }
3354
3355 pub fn has_text_index(&self) -> bool {
3358 self.header.text_index_len > 0
3359 }
3360
3361 pub fn text_index_token_table_len(&self) -> Option<u64> {
3376 self.text_index.token_table_len()
3377 }
3378
3379 pub fn text_search(&self, words: &[&str], prefix: Option<&str>, limit: usize) -> Vec<String> {
3389 text_search_in(self.text_index(), &self.dict, words, prefix, limit)
3390 }
3391
3392 pub fn default_index(&self) -> &GraphIndex {
3394 &self.index
3395 }
3396
3397 pub fn dump(&self, graph: Option<&str>) -> Vec<TermTriple> {
3399 let index = match graph {
3400 None => &self.index,
3401 Some(g) => match self.graph_index(g) {
3402 Some(i) => i,
3403 None => return Vec::new(),
3404 },
3405 };
3406 index
3407 .match_pattern((None, None, None))
3408 .into_iter()
3409 .filter_map(|(s, p, o)| {
3410 Some((
3411 self.dict.subject_term(s)?,
3412 self.dict.predicate_term(p)?,
3413 self.dict.object_term(o)?,
3414 ))
3415 })
3416 .collect()
3417 }
3418
3419 pub fn dump_each<F: FnMut(&str, &str, &str)>(&self, graph: Option<&str>, f: F) {
3435 self.dump_filtered_each(graph, None, None, None, f)
3436 }
3437
3438 pub fn dump_filtered_each<F: FnMut(&str, &str, &str)>(
3509 &self,
3510 graph: Option<&str>,
3511 s: Option<&str>,
3512 p: Option<&str>,
3513 o: Option<&str>,
3514 mut f: F,
3515 ) {
3516 let index = match graph {
3517 None => Some(&self.index),
3518 Some(g) => self.graph_index(g),
3519 };
3520 let Some(index) = index else { return };
3521 let Some(pattern) = self.resolve_query_pattern(s, p, o) else {
3524 return;
3525 };
3526 let dict = &self.dict;
3527 const WINDOW: usize = 4096;
3572 let cap = dict.cache_cap();
3583 let prefetch_window =
3584 cap == u64::MAX || cap >= (WINDOW as u64) * 2 * (DICT_CHUNK_BUDGET as u64);
3585 const PREFETCH_BATCH: usize = 64;
3592 let prefetch_batch = if cap == u64::MAX {
3593 usize::MAX
3594 } else {
3595 PREFETCH_BATCH
3596 };
3597 let mut ids: Vec<(u32, u32, u32)> = Vec::with_capacity(WINDOW);
3598 let mut nodes: Vec<u32> = Vec::with_capacity(WINDOW * 2);
3599 let mut preds: Vec<u32> = Vec::with_capacity(WINDOW);
3600 let mut resolver = crate::dictionary::WindowResolver::default();
3601 let mut flush = |ids: &mut Vec<(u32, u32, u32)>,
3602 resolver: &mut crate::dictionary::WindowResolver,
3603 f: &mut F| {
3604 if ids.is_empty() {
3605 return;
3606 }
3607 if prefetch_window {
3611 nodes.clear();
3612 preds.clear();
3613 for &(s, p, o) in ids.iter() {
3614 nodes.push(dict.subject_node(s));
3615 nodes.push(dict.object_node(o));
3616 preds.push(p);
3617 }
3618 dict.prefetch_terms_batched(&nodes, &preds, prefetch_batch);
3619 }
3620 dict.resolve_window(ids, resolver, |s, p, o| f(s, p, o));
3621 ids.clear();
3622 };
3623 for t in index.scan_iter(pattern) {
3624 ids.push(t);
3625 if ids.len() == WINDOW {
3626 flush(&mut ids, &mut resolver, &mut f);
3627 }
3628 }
3629 flush(&mut ids, &mut resolver, &mut f);
3630 }
3631
3632 pub fn dump_plan(
3640 &self,
3641 graph: Option<&str>,
3642 s: Option<&str>,
3643 p: Option<&str>,
3644 o: Option<&str>,
3645 ) -> DumpPlan {
3646 let index = match graph {
3647 None => Some(&self.index),
3648 Some(g) => self.graph_index(g),
3649 };
3650 let scan = index
3651 .zip(self.resolve_query_pattern(s, p, o))
3652 .map(|(ix, pattern)| ix.scan_plan(pattern));
3653 DumpPlan {
3654 scan,
3655 dictionary_bytes: self.header.dictionary_len,
3656 }
3657 }
3658
3659 pub fn dump_iter(&self, graph: Option<&str>) -> impl Iterator<Item = TermTriple> + '_ {
3677 self.query_iter(graph, None, None, None)
3678 }
3679
3680 pub fn query_iter(
3691 &self,
3692 graph: Option<&str>,
3693 s: Option<&str>,
3694 p: Option<&str>,
3695 o: Option<&str>,
3696 ) -> impl Iterator<Item = TermTriple> + '_ {
3697 let index = match graph {
3698 None => Some(&self.index),
3699 Some(g) => self.graph_index(g),
3700 };
3701 index
3702 .zip(self.resolve_query_pattern(s, p, o))
3703 .into_iter()
3704 .flat_map(|(ix, pattern)| ix.scan_iter(pattern))
3705 .filter_map(move |(s, p, o)| {
3706 Some((
3707 self.dict.subject_term(s)?,
3708 self.dict.predicate_term(p)?,
3709 self.dict.object_term(o)?,
3710 ))
3711 })
3712 }
3713
3714 pub fn query_batch(
3733 &self,
3734 graph: Option<&str>,
3735 s: Option<&str>,
3736 p: Option<&str>,
3737 o: Option<&str>,
3738 cursor: u64,
3739 max_quads: usize,
3740 ) -> (Vec<TermTriple>, u64, bool) {
3741 let index = match graph {
3742 None => &self.index,
3743 Some(g) => match self.graph_index(g) {
3746 Some(i) => i,
3747 None => return (Vec::new(), cursor, true),
3748 },
3749 };
3750 let Some(pattern) = self.resolve_query_pattern(s, p, o) else {
3753 return (Vec::new(), cursor, true);
3754 };
3755 let (ids, next, done) = index.scan_batch(pattern, cursor, max_quads);
3756 if !ids.is_empty() {
3760 let mut nodes = Vec::with_capacity(ids.len() * 2);
3761 let mut preds = Vec::with_capacity(ids.len());
3762 for &(s, p, o) in &ids {
3763 nodes.push(self.dict.subject_node(s));
3764 nodes.push(self.dict.object_node(o));
3765 preds.push(p);
3766 }
3767 self.dict.prefetch_terms(&nodes, &preds);
3768 }
3769 let triples = ids
3770 .into_iter()
3771 .filter_map(|(s, p, o)| {
3772 Some((
3773 self.dict.subject_term(s)?,
3774 self.dict.predicate_term(p)?,
3775 self.dict.object_term(o)?,
3776 ))
3777 })
3778 .collect();
3779 (triples, next, done)
3780 }
3781
3782 pub fn dump_batch(
3808 &self,
3809 graph: Option<&str>,
3810 cursor: u32,
3811 max_quads: usize,
3812 ) -> (Vec<TermTriple>, u32, bool) {
3813 fn next_tile_start(tiles: &[crate::index::Tile], sid: u32) -> Option<u32> {
3818 let i = tiles.partition_point(|t| t.leading_range().1 < sid);
3819 tiles.get(i).map(|t| t.leading_range().0.max(sid))
3820 }
3821
3822 let index = match graph {
3823 None => self.default_index(),
3824 Some(g) => match self.graph_index(g) {
3827 Some(i) => i,
3828 None => return (Vec::new(), cursor, true),
3829 },
3830 };
3831 let tiles = index.tile_sections()[IndexPermutation::Spo.section_index()];
3835 let (Some(first), Some(last)) = (
3836 tiles.first().map(|t| t.leading_range().0),
3837 tiles.last().map(|t| t.leading_range().1),
3838 ) else {
3839 return (Vec::new(), cursor, true); };
3841
3842 let mut sid = cursor.max(first);
3843 let mut probes = max_quads.saturating_mul(4).max(1 << 16);
3844 let mut ids: Vec<(u32, u32, u32)> = Vec::new();
3845 let mut exhausted = false;
3846 while sid <= last && probes > 0 {
3847 probes -= 1;
3848 let hits = index.match_pattern((Some(sid), None, None));
3849 if hits.is_empty() {
3850 match next_tile_start(tiles, sid) {
3854 None => {
3855 exhausted = true;
3856 break;
3857 }
3858 Some(next) if next > sid => {
3859 sid = next;
3860 continue;
3861 }
3862 Some(_) => {}
3863 }
3864 } else {
3865 ids.extend(hits);
3866 }
3867 if sid == u32::MAX {
3868 exhausted = true;
3869 break;
3870 }
3871 sid += 1;
3872 if ids.len() >= max_quads {
3873 break;
3874 }
3875 }
3876 let done = exhausted || sid > last;
3877
3878 let dict = self.dictionary();
3885 if !ids.is_empty() {
3886 let mut nodes = Vec::with_capacity(ids.len() * 2);
3887 let mut preds = Vec::with_capacity(ids.len());
3888 for &(s, p, o) in &ids {
3889 nodes.push(dict.subject_node(s));
3890 nodes.push(dict.object_node(o));
3891 preds.push(p);
3892 }
3893 dict.prefetch_terms(&nodes, &preds);
3894 }
3895 let triples = ids
3896 .into_iter()
3897 .filter_map(|(s, p, o)| {
3898 Some((
3899 dict.subject_term(s)?,
3900 dict.predicate_term(p)?,
3901 dict.object_term(o)?,
3902 ))
3903 })
3904 .collect();
3905 (triples, sid, done)
3906 }
3907
3908 pub fn named_graph_count(&self) -> usize {
3911 self.named_graphs.count()
3912 }
3913
3914 pub fn named_graph_name_at(&self, i: usize) -> Option<&str> {
3917 self.named_graphs.name_at(i, false)
3918 }
3919
3920 pub fn named_graph_at(&self, i: usize) -> Option<(&str, &GraphIndex)> {
3924 self.named_graphs.graph_at(i, false)
3925 }
3926
3927 pub(crate) fn named_graph_name_at_demand(&self, i: usize, exhaustive: bool) -> Option<&str> {
3933 self.named_graphs.name_at(i, exhaustive)
3934 }
3935
3936 pub(crate) fn named_graph_at_demand(
3939 &self,
3940 i: usize,
3941 exhaustive: bool,
3942 ) -> Option<(&str, &GraphIndex)> {
3943 self.named_graphs.graph_at(i, exhaustive)
3944 }
3945
3946 pub fn graph_names(&self) -> Vec<&str> {
3951 (0..self.named_graphs.count())
3952 .filter_map(|i| self.named_graphs.name_at(i, false))
3953 .collect()
3954 }
3955
3956 pub fn graph_index(&self, iri: &str) -> Option<&GraphIndex> {
3958 self.named_graphs.find(iri)
3959 }
3960
3961 pub fn match_ids(
3964 &self,
3965 pattern: (Option<u32>, Option<u32>, Option<u32>),
3966 ) -> Vec<(u32, u32, u32)> {
3967 self.index.match_pattern(pattern)
3968 }
3969
3970 pub fn predicate_pairs(&self, predicate: &str) -> Vec<(u32, u32)> {
3973 let pid = match self.dict.predicate_id(predicate) {
3974 Some(p) => p,
3975 None => return Vec::new(),
3976 };
3977 self.index
3978 .match_pattern((None, Some(pid), None))
3979 .into_iter()
3980 .map(|(s, _p, o)| (self.dict.subject_node(s), self.dict.object_node(o)))
3981 .collect()
3982 }
3983
3984 pub fn open_ranged<R: RangeReader>(reader: &R) -> Result<Self, FileError> {
3988 let head = reader.read_at(0, HEADER_LEN as u64)?;
3989 let header = Header::from_bytes(&head)?;
3990
3991 let dict_bytes = reader.read_at(header.dictionary_offset, header.dictionary_len)?;
3992 let dict = decode_dictionary_container(&dict_bytes, header.dict_codec)?;
3993
3994 let index_bytes = reader.read_at(header.root_dir_offset, header.root_dir_len)?;
3995 let index = decode_index_container(&index_bytes, header.block_codec, header.perms)?;
3996 let index_section_ranges =
3997 decode_index_section_ranges(&index_bytes, header.root_dir_offset, header.perms)?;
3998
3999 let pyramid = PyramidSlot::Resident(if header.pyramid_meta_len > 0 {
4000 let mb = reader.read_at(header.pyramid_meta_offset, header.pyramid_meta_len)?;
4001 Some(
4002 PyramidMeta::decode(&mb)
4003 .map_err(|_| FileError::Container("malformed pyramid meta"))?,
4004 )
4005 } else {
4006 None
4007 });
4008
4009 let text_index_bytes = if header.text_index_len > 0 {
4012 Some(reader.read_at(header.text_index_offset, header.text_index_len)?)
4013 } else {
4014 None
4015 };
4016 let text_index = resident_text_index_slot(text_index_bytes.as_deref(), header.block_codec)?;
4017
4018 let named_graphs = NamedGraphsSlot::Resident(if header.named_graphs_len > 0 {
4019 let nb = reader.read_at(header.named_graphs_offset, header.named_graphs_len)?;
4020 decode_named_graphs(&nb, header.block_codec, header.perms)?
4021 } else {
4022 Vec::new()
4023 });
4024
4025 let tile_ranges =
4029 tile_file_ranges(&index_bytes, header.root_dir_offset, &index_section_ranges);
4030 Ok(Self {
4031 header,
4032 dict,
4033 index,
4034 index_section_ranges,
4035 tile_ranges,
4036 pyramid,
4037 text_index,
4038 named_graphs,
4039 metadata: Vec::new(),
4040 service_client: None,
4041 service_error: std::sync::Mutex::new(None),
4042 spill_threshold: std::sync::atomic::AtomicU64::new(u64::MAX),
4043 })
4044 }
4045
4046 pub fn open_ranged_lazy<R: RangeReader + Send + Sync + 'static>(
4058 reader: R,
4059 ) -> Result<Self, FileError> {
4060 let head = reader.read_at(0, HEADER_LEN as u64)?;
4061 let header = Header::from_bytes(&head)?;
4062 let reader = std::sync::Arc::new(reader);
4063 let read_concurrency = reader.concurrency();
4066
4067 let dict = ranged_chunked_dictionary(&reader, &header, [true; 4])?;
4071
4072 let reader_dyn: std::sync::Arc<dyn RangeReader + Send + Sync> = reader.clone();
4076 let (index, index_section_ranges, tile_ranges) = open_index_container_lazy(
4077 &reader_dyn,
4078 ByteRange {
4079 offset: header.root_dir_offset,
4080 len: header.root_dir_len,
4081 },
4082 header.block_codec,
4083 header.has_tile_synopsis(),
4084 read_concurrency,
4085 header.perms,
4086 )?;
4087
4088 let pyramid = ranged_pyramid_slot(&reader, &header);
4092
4093 let text_index = ranged_text_index_slot(&reader, &header);
4098
4099 let named_graphs = if header.named_graphs_len > 0 {
4106 NamedGraphsSlot::Lazy(LazyNamedGraphs::new(
4107 reader_dyn,
4108 ByteRange {
4109 offset: header.named_graphs_offset,
4110 len: header.named_graphs_len,
4111 },
4112 header.block_codec,
4113 header.has_tile_synopsis(),
4114 read_concurrency,
4115 header.perms,
4116 ))
4117 } else {
4118 NamedGraphsSlot::Resident(Vec::new())
4119 };
4120
4121 let rete = Self {
4122 header,
4123 dict,
4124 index,
4125 index_section_ranges,
4126 tile_ranges,
4127 pyramid,
4128 text_index,
4129 named_graphs,
4130 metadata: Vec::new(),
4131 service_client: None,
4132 service_error: std::sync::Mutex::new(None),
4133 spill_threshold: std::sync::atomic::AtomicU64::new(u64::MAX),
4134 };
4135 #[cfg(not(target_arch = "wasm32"))]
4145 rete.set_memory_budget(Some(DEFAULT_QUERY_BUDGET_MB << 20));
4146 Ok(rete)
4147 }
4148
4149 pub fn index_incomplete(&self) -> bool {
4154 self.index.load_incomplete()
4155 || self.dict.load_incomplete()
4156 || self.named_graphs.load_incomplete()
4157 }
4158
4159 pub fn reset_load_failures(&self) {
4167 self.index.reset_load_failure();
4168 self.dict.reset_load_failure();
4169 self.named_graphs.reset_load_failures();
4170 }
4171
4172 pub fn set_memory_budget(&self, budget: Option<u64>) -> MemoryBudget {
4203 let total = budget.unwrap_or(u64::MAX);
4204 self.set_spill_threshold(budget.map(spill_threshold_for));
4205 let mut split = split_memory_budget(total);
4206 let fixed = if total == u64::MAX {
4207 0
4208 } else {
4209 self.fixed_overhead().total()
4210 };
4211 if split.dict_cache != u64::MAX {
4212 split.dict_cache = split
4213 .dict_cache
4214 .saturating_sub(fixed)
4215 .max(MIN_DICT_CACHE_CAP);
4216 }
4217 self.dict.set_cache_cap(split.dict_cache);
4218 self.index.set_cache_cap(split.tile_cache);
4219 if let NamedGraphsSlot::Lazy(lazy) = &self.named_graphs {
4224 lazy.set_tile_cap(split.tile_cache);
4225 }
4226 if std::env::var("RETE_OPEN_DEBUG").is_ok() {
4227 eprintln!(
4228 "[memory-budget] total={} dict_cache={} tile_cache={} block_cache={} fixed={:.1}MiB",
4229 if total == u64::MAX {
4230 "unlimited".to_string()
4231 } else {
4232 format!("{}MiB", total >> 20)
4233 },
4234 fmt_cap(split.dict_cache),
4235 fmt_cap(split.tile_cache),
4236 fmt_cap(split.block_cache),
4237 fixed as f64 / (1u64 << 20) as f64,
4238 );
4239 }
4240 split
4241 }
4242
4243 pub fn spill_threshold(&self) -> u64 {
4250 self.spill_threshold
4251 .load(std::sync::atomic::Ordering::Relaxed)
4252 }
4253
4254 pub fn set_spill_threshold(&self, bytes: Option<u64>) {
4258 self.spill_threshold.store(
4259 bytes.unwrap_or(u64::MAX),
4260 std::sync::atomic::Ordering::Relaxed,
4261 );
4262 }
4263
4264 pub fn dict_cache_stats(&self) -> crate::chunk_cache::CacheStats {
4268 self.dict.cache_stats()
4269 }
4270
4271 pub fn fixed_overhead(&self) -> FixedOverhead {
4275 FixedOverhead {
4276 dict_directory: self.dict.directory_bytes(),
4277 index_directory: self.index.directory_bytes(),
4278 named_graphs: self.named_graphs.directory_bytes(),
4279 }
4280 }
4281
4282 pub fn index_cache_stats(&self) -> crate::chunk_cache::CacheStats {
4284 self.index.cache_stats()
4285 }
4286
4287 pub fn dict_cache_cap(&self) -> u64 {
4289 self.dict.cache_cap()
4290 }
4291
4292 pub fn release_named_graph(&mut self, iri: &str) {
4305 if let NamedGraphsSlot::Lazy(lazy) = &mut self.named_graphs {
4306 lazy.release(iri);
4307 }
4308 }
4309
4310 fn resolve_query_pattern(
4311 &self,
4312 s: Option<&str>,
4313 p: Option<&str>,
4314 o: Option<&str>,
4315 ) -> Option<Pattern> {
4316 let sid = match s {
4317 Some(t) => match self.dict.subject_id(t) {
4318 Some(id) => Some(id),
4319 None => return None,
4320 },
4321 None => None,
4322 };
4323 let pid = match p {
4324 Some(t) => match self.dict.predicate_id(t) {
4325 Some(id) => Some(id),
4326 None => return None,
4327 },
4328 None => None,
4329 };
4330 let oid = match o {
4331 Some(t) => match self.dict.object_id(t) {
4332 Some(id) => Some(id),
4333 None => return None,
4334 },
4335 None => None,
4336 };
4337 Some((sid, pid, oid))
4338 }
4339
4340 pub fn query_with_provenance(
4344 &self,
4345 s: Option<&str>,
4346 p: Option<&str>,
4347 o: Option<&str>,
4348 ) -> Vec<TripleProvenance> {
4349 let pattern = match self.resolve_query_pattern(s, p, o) {
4350 Some(pattern) => pattern,
4351 None => return Vec::new(),
4352 };
4353
4354 let index_permutation = GraphIndex::best_permutation_in(self.header.perms, pattern);
4355 let dictionary_range = ByteRange {
4356 offset: self.header.dictionary_offset,
4357 len: self.header.dictionary_len,
4358 };
4359 let index_range = ByteRange {
4360 offset: self.header.root_dir_offset,
4361 len: self.header.root_dir_len,
4362 };
4363 let index_section_range = self.index_section_ranges[index_permutation.section_index()];
4364 let pyramid_range = (self.header.pyramid_meta_len > 0).then_some(ByteRange {
4365 offset: self.header.pyramid_meta_offset,
4366 len: self.header.pyramid_meta_len,
4367 });
4368
4369 let tiles = &self.tile_ranges[index_permutation.section_index()];
4370 self.index
4371 .match_pattern(pattern)
4372 .into_iter()
4373 .filter_map(|(s, p, o)| {
4374 let terms = (
4375 self.dict.subject_term(s)?,
4376 self.dict.predicate_term(p)?,
4377 self.dict.object_term(o)?,
4378 );
4379 let a = index_permutation.forward((s, p, o)).0;
4382 let ti = tiles.partition_point(|&(_, max_a, _)| max_a < a);
4383 let (tile, tile_range) = match tiles.get(ti) {
4384 Some(&(min_a, _, range)) if min_a <= a => (
4385 Some(format!("{}/{ti}", index_permutation.name())),
4386 Some(range),
4387 ),
4388 _ => (None, None),
4389 };
4390 Some(TripleProvenance {
4391 terms,
4392 ids: (s, p, o),
4393 graph: None,
4394 matched_pattern: pattern,
4395 index_permutation,
4396 dictionary_range,
4397 index_range,
4398 index_section_range,
4399 pyramid_range,
4400 tile,
4401 tile_range,
4402 })
4403 })
4404 .collect()
4405 }
4406
4407 pub fn query(&self, s: Option<&str>, p: Option<&str>, o: Option<&str>) -> Vec<TermTriple> {
4411 self.query_with_provenance(s, p, o)
4412 .into_iter()
4413 .map(|m| m.terms)
4414 .collect()
4415 }
4416
4417 pub fn query_in_graph(
4425 &self,
4426 graph: Option<&str>,
4427 s: Option<&str>,
4428 p: Option<&str>,
4429 o: Option<&str>,
4430 ) -> Vec<TermTriple> {
4431 let pattern = match self.resolve_query_pattern(s, p, o) {
4432 Some(pattern) => pattern,
4433 None => return Vec::new(),
4434 };
4435 let index = match graph {
4436 None => &self.index,
4437 Some(g) => match self.graph_index(g) {
4438 Some(i) => i,
4439 None => return Vec::new(),
4440 },
4441 };
4442 let ids = index.match_pattern(pattern);
4443 if !ids.is_empty() {
4451 let mut nodes = Vec::with_capacity(ids.len() * 2);
4452 let mut preds = Vec::with_capacity(ids.len());
4453 for &(s, p, o) in &ids {
4454 nodes.push(self.dict.subject_node(s));
4455 nodes.push(self.dict.object_node(o));
4456 preds.push(p);
4457 }
4458 self.dict.prefetch_terms(&nodes, &preds);
4459 }
4460 ids.into_iter()
4461 .filter_map(|(s, p, o)| {
4462 Some((
4463 self.dict.subject_term(s)?,
4464 self.dict.predicate_term(p)?,
4465 self.dict.object_term(o)?,
4466 ))
4467 })
4468 .collect()
4469 }
4470
4471 pub fn query_quads(
4476 &self,
4477 s: Option<&str>,
4478 p: Option<&str>,
4479 o: Option<&str>,
4480 ) -> Vec<(TermTriple, Option<String>)> {
4481 let mut out: Vec<(TermTriple, Option<String>)> = self
4482 .query_in_graph(None, s, p, o)
4483 .into_iter()
4484 .map(|t| (t, None))
4485 .collect();
4486 for i in 0..self.named_graphs.count() {
4487 let Some(iri) = self.named_graphs.name_at(i, true) else {
4490 continue;
4491 };
4492 for triple in self.query_in_graph(Some(iri), s, p, o) {
4493 out.push((triple, Some(iri.to_string())));
4494 }
4495 }
4496 out
4497 }
4498
4499 pub fn query_ranged<R: RangeReader>(
4506 reader: &R,
4507 s: Option<&str>,
4508 p: Option<&str>,
4509 o: Option<&str>,
4510 ) -> Result<Vec<TermTriple>, FileError> {
4511 let routed = match route_pattern(reader, s, p, o)? {
4512 Some(routed) => routed,
4513 None => return Ok(Vec::new()),
4514 };
4515 let matches = fetch_routed_matches(reader, &routed)?;
4516 Ok(matches
4517 .into_iter()
4518 .filter_map(|(s, p, o)| {
4519 Some((
4520 routed.dict.subject_term(s)?,
4521 routed.dict.predicate_term(p)?,
4522 routed.dict.object_term(o)?,
4523 ))
4524 })
4525 .collect())
4526 }
4527
4528 pub fn route_pattern_ranged<R: RangeReader>(
4532 reader: &R,
4533 s: Option<&str>,
4534 p: Option<&str>,
4535 o: Option<&str>,
4536 ) -> Result<bool, FileError> {
4537 Ok(route_pattern(reader, s, p, o)?.is_some())
4538 }
4539}
4540
4541struct RoutedPattern {
4544 dict: Dictionary,
4545 pattern: Pattern,
4546 permutation: IndexPermutation,
4547 header: Header,
4548 section: ByteRange,
4550}
4551
4552fn route_pattern<R: RangeReader>(
4555 reader: &R,
4556 s: Option<&str>,
4557 p: Option<&str>,
4558 o: Option<&str>,
4559) -> Result<Option<RoutedPattern>, FileError> {
4560 let head = reader.read_at(0, HEADER_LEN as u64)?;
4561 let header = Header::from_bytes(&head)?;
4562
4563 let dict_bytes = reader.read_at(header.dictionary_offset, header.dictionary_len)?;
4564 let dict = decode_dictionary_container(&dict_bytes, header.dict_codec)?;
4565
4566 let Some(pattern) = resolve_query_pattern(&dict, s, p, o) else {
4567 return Ok(None);
4568 };
4569 let permutation = GraphIndex::best_permutation_in(header.perms, pattern);
4570 let section = locate_container_section_ranged(
4571 reader,
4572 header.root_dir_offset,
4573 header.root_dir_len,
4574 header
4575 .perms
4576 .position(permutation)
4577 .ok_or(FileError::Container("routed to an absent permutation"))?,
4578 header.perms.len() as u64,
4579 )?;
4580 Ok(Some(RoutedPattern {
4581 dict,
4582 pattern,
4583 permutation,
4584 header,
4585 section,
4586 }))
4587}
4588
4589fn fetch_routed_matches<R: RangeReader>(
4594 reader: &R,
4595 routed: &RoutedPattern,
4596) -> Result<Vec<Triple>, FileError> {
4597 let dir = read_tile_directory_ranged(reader, routed.section)?;
4598 let [pa, _, _] = routed.permutation.order_pattern(routed.pattern);
4599 let codec = routed.header.block_codec;
4600 let mut out = Vec::new();
4601 match pa {
4602 Some(a) => {
4605 for e in dir.iter().filter(|e| e.min_a <= a && a <= e.max_a) {
4606 let bytes = reader.read_at(routed.section.offset + e.start, e.end - e.start)?;
4607 let tile = decompress(codec, &bytes)?;
4608 out.extend(GraphIndex::match_serialized_block(
4609 &tile,
4610 routed.permutation,
4611 routed.pattern,
4612 ));
4613 }
4614 }
4615 None => {
4618 if let (Some(first), Some(last)) = (dir.first(), dir.last()) {
4619 let base = first.start;
4620 let body = reader.read_at(routed.section.offset + base, last.end - base)?;
4621 for e in &dir {
4622 let tile = decompress(
4623 codec,
4624 &body[(e.start - base) as usize..(e.end - base) as usize],
4625 )?;
4626 out.extend(GraphIndex::match_serialized_block(
4627 &tile,
4628 routed.permutation,
4629 routed.pattern,
4630 ));
4631 }
4632 }
4633 }
4634 }
4635 out.sort_unstable();
4636 Ok(out)
4637}
4638
4639fn resolve_query_pattern(
4640 dict: &Dictionary,
4641 s: Option<&str>,
4642 p: Option<&str>,
4643 o: Option<&str>,
4644) -> Option<Pattern> {
4645 let sid = match s {
4646 Some(t) => Some(dict.subject_id(t)?),
4647 None => None,
4648 };
4649 let pid = match p {
4650 Some(t) => Some(dict.predicate_id(t)?),
4651 None => None,
4652 };
4653 let oid = match o {
4654 Some(t) => Some(dict.object_id(t)?),
4655 None => None,
4656 };
4657 Some((sid, pid, oid))
4658}
4659
4660fn ranged_chunked_dictionary<R: RangeReader + Send + Sync + 'static>(
4675 reader: &std::sync::Arc<R>,
4676 header: &Header,
4677 want: [bool; 4],
4678) -> Result<Dictionary, FileError> {
4679 let cache = crate::chunk_cache::ChunkCache::unlimited_arc();
4682 let mut dict_sections: Vec<crate::dict::ChunkedSection> = Vec::with_capacity(4);
4683 for si in 0..4 {
4684 if !want[si] {
4685 dict_sections.push(crate::dict::ChunkedSection::from_parts(
4686 crate::dict::SectionMeta {
4687 term_count: 0,
4688 restart_interval: 1,
4689 restart_offsets: Vec::new(),
4690 },
4691 Vec::new(),
4692 None,
4693 cache.clone(),
4694 si as u8,
4695 ));
4696 continue;
4697 }
4698 let section = locate_container_section_ranged(
4699 reader.as_ref(),
4700 header.dictionary_offset,
4701 header.dictionary_len,
4702 si,
4703 4,
4704 )?;
4705 let (meta, entries) = read_dict_dir_ranged(reader.as_ref(), section)?;
4706 let ranges: Vec<ByteRange> = entries
4707 .iter()
4708 .map(|e| ByteRange {
4709 offset: section.offset + e.start,
4710 len: (e.end - e.start),
4711 })
4712 .collect();
4713 let chunks: Vec<crate::dict::SectionChunk> = entries
4714 .into_iter()
4715 .map(|e| crate::dict::SectionChunk::new(e.first_run, e.key, e.body_start))
4716 .collect();
4717 let chunk_reader = reader.clone();
4718 let codec = header.dict_codec;
4719 let loader_ranges = ranges.clone();
4720 let loader: crate::dict::ChunkLoader = Box::new(move |ci| {
4721 let range = loader_ranges.get(ci)?;
4722 let bytes = chunk_reader.read_at(range.offset, range.len).ok()?;
4723 decompress(codec, &bytes).ok()
4724 });
4725 let bulk_reader = reader.clone();
4728 let bulk: crate::dict::ChunkBulkLoader = Box::new(move |cis| {
4729 let want: Option<Vec<ByteRange>> =
4730 cis.iter().map(|&ci| ranges.get(ci).copied()).collect();
4731 let blobs = read_coalesced(bulk_reader.as_ref(), &want?, DICT_COALESCE_GAP)?;
4732 blobs.iter().map(|b| decompress(codec, b).ok()).collect()
4733 });
4734 dict_sections.push(
4735 crate::dict::ChunkedSection::from_parts(
4736 meta,
4737 chunks,
4738 Some(loader),
4739 cache.clone(),
4740 si as u8,
4741 )
4742 .with_bulk_loader(bulk),
4743 );
4744 }
4745 let dict_arr: [crate::dict::ChunkedSection; 4] = dict_sections
4746 .try_into()
4747 .map_err(|_| FileError::Container("expected 4 dictionary sections"))?;
4748 Ok(Dictionary::from_chunked_sections(dict_arr))
4749}
4750
4751fn ranged_pyramid_slot<R: RangeReader + Send + Sync + 'static>(
4755 reader: &std::sync::Arc<R>,
4756 header: &Header,
4757) -> PyramidSlot {
4758 if header.pyramid_meta_len == 0 {
4759 return PyramidSlot::Resident(None);
4760 }
4761 let pyr_reader = reader.clone();
4762 let pyr_off = header.pyramid_meta_offset;
4763 let pyr_len = header.pyramid_meta_len;
4764 PyramidSlot::Lazy {
4765 loader: Box::new(move || {
4766 let mb = pyr_reader.read_at(pyr_off, pyr_len).ok()?;
4767 PyramidMeta::decode(&mb).ok()
4768 }),
4769 cell: std::sync::OnceLock::new(),
4770 }
4771}
4772
4773fn ranged_text_index_slot<R: RangeReader + Send + Sync + 'static>(
4778 reader: &std::sync::Arc<R>,
4779 header: &Header,
4780) -> TextIndexSlot {
4781 if header.text_index_len == 0 {
4782 return TextIndexSlot::Resident {
4783 index: None,
4784 token_table_len: None,
4785 };
4786 }
4787 let ti_reader = reader.clone();
4788 let probe_reader = reader.clone();
4789 let ti_off = header.text_index_offset;
4790 let ti_len = header.text_index_len;
4791 let codec = header.block_codec;
4792 TextIndexSlot::Lazy {
4793 loader: Box::new(move || {
4794 let prefix_len = read_token_table_len(&*ti_reader, ti_off, ti_len)?;
4798 let prefix = ti_reader.read_at(ti_off, prefix_len).ok()?;
4799 let postings_base = crate::text_index::TextIndex::postings_base(&prefix)? as u64;
4800 let postings_abs = ti_off + postings_base;
4801 let pr = ti_reader.clone();
4802 let posting_loader =
4803 Box::new(move |off: u64, len: u64| pr.read_at(postings_abs + off, len).ok());
4804 crate::text_index::TextIndex::from_token_table(&prefix, codec, posting_loader).ok()
4805 }),
4806 cell: std::sync::OnceLock::new(),
4807 token_table: Box::new(move || read_token_table_len(&*probe_reader, ti_off, ti_len)),
4810 token_table_cell: std::sync::OnceLock::new(),
4811 }
4812}
4813
4814fn text_search_in(
4817 ti: Option<&crate::text_index::TextIndex>,
4818 dict: &Dictionary,
4819 words: &[&str],
4820 prefix: Option<&str>,
4821 limit: usize,
4822) -> Vec<String> {
4823 let Some(ti) = ti else {
4824 return Vec::new();
4825 };
4826 let mut acc: Option<Vec<u32>> = None;
4830 if let Some(p) = prefix {
4831 acc = Some(ti.prefix(&p.to_lowercase()));
4832 }
4833 for w in words {
4834 for tok in crate::text_index::tokenize(w) {
4835 let posting = ti.lookup(&tok);
4836 acc = Some(match acc {
4837 Some(a) => intersect_sorted(&a, &posting),
4838 None => posting,
4839 });
4840 if acc.as_ref().is_some_and(|a| a.is_empty()) {
4841 return Vec::new();
4842 }
4843 }
4844 }
4845 let ids = acc.unwrap_or_default();
4846 let mut out = Vec::with_capacity(if limit > 0 {
4847 limit.min(ids.len())
4848 } else {
4849 ids.len()
4850 });
4851 for id in ids {
4852 if let Some(iri) = dict.subject_term(id) {
4853 out.push(iri);
4854 if limit > 0 && out.len() >= limit {
4855 break;
4856 }
4857 }
4858 }
4859 out
4860}
4861
4862pub struct SearchView {
4880 header: Header,
4881 dict: Dictionary,
4882 pyramid: PyramidSlot,
4883 text_index: TextIndexSlot,
4884}
4885
4886impl SearchView {
4887 pub fn open_ranged<R: RangeReader + Send + Sync + 'static>(
4891 reader: R,
4892 ) -> Result<Self, FileError> {
4893 let head = reader.read_at(0, HEADER_LEN as u64)?;
4894 let header = Header::from_bytes(&head)?;
4895 let reader = std::sync::Arc::new(reader);
4896 let dict = ranged_chunked_dictionary(&reader, &header, [true, true, false, false])?;
4900 let pyramid = ranged_pyramid_slot(&reader, &header);
4901 let text_index = ranged_text_index_slot(&reader, &header);
4902 Ok(Self {
4903 header,
4904 dict,
4905 pyramid,
4906 text_index,
4907 })
4908 }
4909
4910 pub fn header(&self) -> &Header {
4912 &self.header
4913 }
4914
4915 pub fn has_text_index(&self) -> bool {
4917 self.header.text_index_len > 0
4918 }
4919
4920 pub fn has_pyramid(&self) -> bool {
4923 self.header.pyramid_meta_len > 0
4924 }
4925
4926 pub fn text_search(&self, words: &[&str], prefix: Option<&str>, limit: usize) -> Vec<String> {
4929 text_search_in(self.text_index.index(), &self.dict, words, prefix, limit)
4930 }
4931
4932 pub fn text_index_token_table_len(&self) -> Option<u64> {
4936 self.text_index.token_table_len()
4937 }
4938
4939 pub fn prefix_search(&self, prefix: &str, limit: usize) -> Vec<(String, String)> {
4942 let pyr = match &self.pyramid {
4943 PyramidSlot::Resident(p) => p.as_ref(),
4944 PyramidSlot::Lazy { loader, cell } => cell.get_or_init(loader).as_ref(),
4945 };
4946 let Some(pyr) = pyr else {
4947 return Vec::new();
4948 };
4949 pyr.prefix_search(prefix, limit)
4950 .into_iter()
4951 .filter_map(|e| {
4952 self.dict
4953 .subject_term(e.subject)
4954 .map(|iri| (e.label.clone(), iri))
4955 })
4956 .collect()
4957 }
4958}
4959
4960#[must_use]
4965pub struct SummaryView {
4966 pub round: u32,
4967 pub summary: Vec<SuperEdge>,
4968 pub class_hierarchy: Vec<ClassNode>,
4970 pub level_rollups: Vec<LevelRollup>,
4972 pub level_links: Vec<LevelLinks>,
4974 pub descriptors: Vec<CommunityDescriptor>,
4976 pub subclass_cycles: Vec<Vec<String>>,
4978 pub disjoint_pairs: Vec<(String, String)>,
4980 pub equivalent_pairs: Vec<(String, String)>,
4982 dict: Dictionary,
4983}
4984
4985impl SummaryView {
4986 pub fn open_ranged<R: RangeReader>(reader: &R) -> Result<Option<Self>, FileError> {
4988 let head = reader.read_at(0, HEADER_LEN as u64)?;
4989 let header = Header::from_bytes(&head)?;
4990 if header.pyramid_meta_len == 0 {
4991 return Ok(None);
4992 }
4993
4994 let dict_bytes = reader.read_at(header.dictionary_offset, header.dictionary_len)?;
4995 let dict = decode_dictionary_container(&dict_bytes, header.dict_codec)?;
4996
4997 let mb = reader.read_at(header.pyramid_meta_offset, header.pyramid_meta_len)?;
4998 let meta =
4999 PyramidMeta::decode(&mb).map_err(|_| FileError::Container("malformed pyramid meta"))?;
5000
5001 Ok(Some(SummaryView {
5002 round: meta.round,
5003 summary: meta.summary,
5004 class_hierarchy: meta.class_hierarchy,
5005 level_rollups: meta.level_rollups,
5006 level_links: meta.level_links,
5007 descriptors: meta.descriptors,
5008 subclass_cycles: meta.subclass_cycles,
5009 disjoint_pairs: meta.disjoint_pairs,
5010 equivalent_pairs: meta.equivalent_pairs,
5011 dict,
5012 }))
5013 }
5014
5015 pub fn level_count(&self) -> usize {
5017 self.level_rollups.len()
5018 }
5019
5020 pub fn level_rollup(&self, k: usize) -> Option<&LevelRollup> {
5023 self.level_rollups.get(k)
5024 }
5025
5026 pub fn predicate_term(&self, id: u32) -> Option<String> {
5028 self.dict.predicate_term(id)
5029 }
5030
5031 pub fn predicate_total(&self, predicate: &str) -> u32 {
5034 match self.dict.predicate_id(predicate) {
5035 Some(pid) => self
5036 .summary
5037 .iter()
5038 .filter(|e| e.predicate == pid)
5039 .map(|e| e.count)
5040 .sum(),
5041 None => 0,
5042 }
5043 }
5044
5045 pub fn predicate_totals(&self) -> Vec<(String, u32)> {
5047 let mut by_pred: std::collections::BTreeMap<u32, u32> = std::collections::BTreeMap::new();
5048 for e in &self.summary {
5049 *by_pred.entry(e.predicate).or_default() += e.count;
5050 }
5051 let mut out: Vec<(String, u32)> = by_pred
5052 .into_iter()
5053 .filter_map(|(pid, c)| self.dict.predicate_term(pid).map(|t| (t, c)))
5054 .collect();
5055 out.sort_by(|a, b| b.1.cmp(&a.1).then_with(|| a.0.cmp(&b.0)));
5056 out
5057 }
5058
5059 pub fn community_count(&self) -> usize {
5061 let mut comms = std::collections::BTreeSet::new();
5062 for e in &self.summary {
5063 comms.insert(e.s_comm);
5064 comms.insert(e.o_comm);
5065 }
5066 comms.len()
5067 }
5068
5069 pub fn tbox_coherence(&self) -> Vec<crate::reason::Inconsistency> {
5084 schema_coherence(
5085 &self.class_hierarchy,
5086 &self.subclass_cycles,
5087 &self.disjoint_pairs,
5088 &self.equivalent_pairs,
5089 )
5090 }
5091
5092 pub fn tbox_is_coherent(&self) -> bool {
5095 self.tbox_coherence().is_empty()
5096 }
5097}
5098
5099pub fn schema_coherence(
5105 class_hierarchy: &[ClassNode],
5106 subclass_cycles: &[Vec<String>],
5107 disjoint_pairs: &[(String, String)],
5108 equivalent_pairs: &[(String, String)],
5109) -> Vec<crate::reason::Inconsistency> {
5110 use crate::reason::Inconsistency;
5111 use std::collections::{BTreeMap, BTreeSet, VecDeque};
5112 const MAX_REACH: usize = 100_000;
5113
5114 let mut out: Vec<Inconsistency> = Vec::new();
5115
5116 for cyc in subclass_cycles {
5117 let detail = if cyc.len() == 1 {
5118 format!("{} is rdfs:subClassOf itself (a cycle)", cyc[0])
5119 } else {
5120 format!(
5121 "classes {{{}}} are mutually rdfs:subClassOf (a cycle)",
5122 cyc.join(", ")
5123 )
5124 };
5125 out.push(Inconsistency {
5126 kind: "subclass-cycle",
5127 detail,
5128 });
5129 }
5130
5131 if !disjoint_pairs.is_empty() {
5132 let mut adj: BTreeMap<&str, Vec<&str>> = BTreeMap::new();
5134 for n in class_hierarchy {
5135 let e = adj.entry(n.class.as_str()).or_default();
5136 for p in &n.parents {
5137 e.push(p.as_str());
5138 }
5139 }
5140 for (a, b) in equivalent_pairs {
5141 adj.entry(a.as_str()).or_default().push(b.as_str());
5142 adj.entry(b.as_str()).or_default().push(a.as_str());
5143 }
5144
5145 let mut focuses: BTreeSet<&str> =
5147 class_hierarchy.iter().map(|n| n.class.as_str()).collect();
5148 for (a, b) in disjoint_pairs.iter().chain(equivalent_pairs) {
5149 focuses.insert(a.as_str());
5150 focuses.insert(b.as_str());
5151 }
5152
5153 let mut seen: BTreeSet<&str> = BTreeSet::new();
5154 for &c in &focuses {
5155 let mut reach: BTreeSet<&str> = BTreeSet::new();
5157 let mut q: VecDeque<&str> = VecDeque::new();
5158 reach.insert(c);
5159 q.push_back(c);
5160 while let Some(x) = q.pop_front() {
5161 if reach.len() > MAX_REACH {
5162 break;
5163 }
5164 if let Some(ns) = adj.get(x) {
5165 for &p in ns {
5166 if reach.insert(p) {
5167 q.push_back(p);
5168 }
5169 }
5170 }
5171 }
5172 for (x, y) in disjoint_pairs {
5173 if reach.contains(x.as_str()) && reach.contains(y.as_str()) && seen.insert(c) {
5174 out.push(Inconsistency {
5175 kind: "unsatisfiable-class",
5176 detail: format!(
5177 "{c} is a subclass of both {x} and {y}, which are \
5178 owl:disjointWith — no individual can be a {c}"
5179 ),
5180 });
5181 break;
5182 }
5183 }
5184 }
5185 }
5186
5187 out.sort_by(|a, b| (a.kind, &a.detail).cmp(&(b.kind, &b.detail)));
5188 out
5189}
5190
5191pub fn read_schema_coherence_ranged<R: RangeReader>(
5199 reader: &R,
5200) -> Result<Option<Vec<crate::reason::Inconsistency>>, FileError> {
5201 let head = reader.read_at(0, HEADER_LEN as u64)?;
5202 let header = Header::from_bytes(&head)?;
5203 if header.pyramid_meta_len == 0 {
5204 return Ok(None);
5205 }
5206 if header.schema_meta_len > 0 && (header.schema_meta_len as u64) <= header.pyramid_meta_len {
5210 let off =
5211 header.pyramid_meta_offset + header.pyramid_meta_len - header.schema_meta_len as u64;
5212 let block = reader.read_at(off, header.schema_meta_len as u64)?;
5213 let (hierarchy, cycles, disjoint, equivalent) = crate::meta::decode_schema_block(&block)
5214 .map_err(|_| FileError::Container("malformed schema block"))?;
5215 return Ok(Some(schema_coherence(
5216 &hierarchy,
5217 &cycles,
5218 &disjoint,
5219 &equivalent,
5220 )));
5221 }
5222 let mb = reader.read_at(header.pyramid_meta_offset, header.pyramid_meta_len)?;
5224 let meta =
5225 PyramidMeta::decode(&mb).map_err(|_| FileError::Container("malformed pyramid meta"))?;
5226 Ok(Some(schema_coherence(
5227 &meta.class_hierarchy,
5228 &meta.subclass_cycles,
5229 &meta.disjoint_pairs,
5230 &meta.equivalent_pairs,
5231 )))
5232}
5233
5234#[allow(clippy::type_complexity)]
5242pub fn read_schema_summary_ranged<R: RangeReader>(
5243 reader: &R,
5244) -> Result<Option<(Vec<(String, u64)>, Vec<(String, String, String, u64)>)>, FileError> {
5245 let head = reader.read_at(0, HEADER_LEN as u64)?;
5246 let header = Header::from_bytes(&head)?;
5247 if header.pyramid_meta_len == 0 {
5248 return Ok(None);
5249 }
5250 if header.schema_meta_len > 0 && (header.schema_meta_len as u64) <= header.pyramid_meta_len {
5251 let off =
5252 header.pyramid_meta_offset + header.pyramid_meta_len - header.schema_meta_len as u64;
5253 let block = reader.read_at(off, header.schema_meta_len as u64)?;
5254 let summary = crate::meta::decode_schema_block_summary(&block)
5255 .map_err(|_| FileError::Container("malformed schema block"))?;
5256 return Ok(Some(summary));
5257 }
5258 let mb = reader.read_at(header.pyramid_meta_offset, header.pyramid_meta_len)?;
5260 let meta =
5261 PyramidMeta::decode(&mb).map_err(|_| FileError::Container("malformed pyramid meta"))?;
5262 if meta.level_rollups.is_empty() && meta.level_links.is_empty() {
5263 return Ok(None);
5264 }
5265 let classes = meta
5266 .level_rollups
5267 .iter()
5268 .max_by_key(|r| r.depth)
5269 .map(|r| r.classes.clone())
5270 .unwrap_or_default();
5271 let relations = meta
5272 .level_links
5273 .iter()
5274 .max_by_key(|l| l.depth)
5275 .map(|l| {
5276 l.links
5277 .iter()
5278 .map(|c| {
5279 (
5280 c.s_class.clone(),
5281 c.predicate.clone(),
5282 c.o_class.clone(),
5283 c.count,
5284 )
5285 })
5286 .collect()
5287 })
5288 .unwrap_or_default();
5289 Ok(Some((classes, relations)))
5290}
5291
5292#[cfg(test)]
5297pub(crate) fn dict_chunk_keys_for_test(image: &[u8]) -> Vec<Vec<Vec<u8>>> {
5298 let header = Header::from_bytes(&image[..HEADER_LEN]).unwrap();
5299 let s = header.dictionary_offset as usize;
5300 let e = s + header.dictionary_len as usize;
5301 decode_container(&image[s..e], CODEC_NONE)
5302 .unwrap()
5303 .iter()
5304 .map(|payload| {
5305 let (_meta, entries) = parse_chunked_dict_dir(payload, payload.len() as u64).unwrap();
5306 entries.into_iter().map(|c| c.key).collect()
5307 })
5308 .collect()
5309}
5310
5311#[cfg(test)]
5312mod tests {
5313 use super::*;
5314 use crate::dictionary::DictionaryBuilder;
5315 use crate::index::GraphIndexBuilder;
5316
5317 #[cfg(not(feature = "compression"))]
5320 #[test]
5321 fn ruzstd_decodes_a_zstd_frame() {
5322 let data: Vec<u8> = (0..50_000u32)
5323 .flat_map(|i| (i % 251).to_le_bytes())
5324 .collect();
5325 let frame = ruzstd::encoding::compress_to_vec(
5326 &data[..],
5327 ruzstd::encoding::CompressionLevel::Fastest,
5328 );
5329 assert!(frame.len() < data.len());
5330 assert_eq!(decompress(CODEC_ZSTD, &frame).unwrap(), data);
5331 }
5332
5333 #[test]
5339 fn three_permutation_file_round_trips_every_open_path() {
5340 use crate::index::PermSet;
5341 use crate::reader::SliceReader;
5342
5343 let build = |perms: PermSet| {
5344 let mut db = crate::DictionaryBuilder::new();
5345 let mut triples = Vec::new();
5346 for i in 0..60u32 {
5347 let (s, p, o) = (
5348 format!("<http://ex/s{}>", i % 11),
5349 format!("<http://ex/p{}>", i % 3),
5350 format!("<http://ex/o{}>", i % 7),
5351 );
5352 db.observe(&s, &p, &o);
5353 triples.push((s, p, o));
5354 }
5355 let dict = db.build();
5356 let ids: Vec<(u32, u32, u32)> = triples
5357 .iter()
5358 .map(|(s, p, o)| dict.encode(s, p, o).unwrap())
5359 .collect();
5360 let def = GraphIndexBuilder::from_triples(ids.clone())
5361 .with_tile_budget(96)
5362 .with_perms(perms)
5363 .build();
5364 let named = GraphIndexBuilder::from_triples(ids)
5365 .with_tile_budget(96)
5366 .with_perms(perms)
5367 .build();
5368 write_dataset(
5369 &dict,
5370 &def,
5371 &[("<http://ex/g>".to_string(), named)],
5372 true,
5373 &[],
5374 0,
5375 )
5376 };
5377
5378 let six = build(PermSet::ALL);
5379 let three = build(PermSet::CORE);
5380 assert!(three.len() < six.len());
5381 assert_eq!(Header::from_bytes(&three).unwrap().perms, PermSet::CORE);
5382 assert_eq!(Header::from_bytes(&six).unwrap().perms, PermSet::ALL);
5383
5384 let shapes: Vec<(Option<&str>, Option<&str>, Option<&str>)> = vec![
5385 (None, None, None),
5386 (Some("<http://ex/s3>"), None, None),
5387 (None, Some("<http://ex/p1>"), None),
5388 (None, None, Some("<http://ex/o5>")),
5389 (Some("<http://ex/s3>"), Some("<http://ex/p1>"), None),
5390 (Some("<http://ex/s3>"), None, Some("<http://ex/o5>")),
5391 (None, Some("<http://ex/p1>"), Some("<http://ex/o5>")),
5392 (
5393 Some("<http://ex/s3>"),
5394 Some("<http://ex/p1>"),
5395 Some("<http://ex/o5>"),
5396 ),
5397 ];
5398
5399 for (image, tag) in [(&six, "six"), (&three, "three")] {
5400 let resident = Rete::open(image).unwrap();
5401 let leaked: &'static [u8] = Box::leak(image.clone().into_boxed_slice());
5402 let ranged = Rete::open_ranged(&SliceReader::new(leaked)).unwrap();
5403 let lazy = Rete::open_ranged_lazy(SliceReader::new(leaked)).unwrap();
5404 for (s, p, o) in &shapes {
5405 let want: Vec<_> = {
5406 let mut v = Rete::open(&six).unwrap().query(*s, *p, *o);
5407 v.sort();
5408 v
5409 };
5410 for (got, path) in [
5411 (resident.query(*s, *p, *o), "resident"),
5412 (ranged.query(*s, *p, *o), "ranged"),
5413 (lazy.query(*s, *p, *o), "lazy"),
5414 ] {
5415 let mut got = got;
5416 got.sort();
5417 assert_eq!(got, want, "{tag}/{path} disagreed on {:?}", (s, p, o));
5418 }
5419 let mut g = resident.query_in_graph(Some("<http://ex/g>"), *s, *p, *o);
5421 g.sort();
5422 assert_eq!(g, want, "{tag}/named disagreed on {:?}", (s, p, o));
5423 }
5424 }
5425
5426 let leaked: &'static [u8] = Box::leak(three.clone().into_boxed_slice());
5430 let reader = SliceReader::new(leaked);
5431 let routed = Rete::query_ranged(&reader, None, None, Some("<http://ex/o5>")).unwrap();
5432 let mut want = Rete::open(&six)
5433 .unwrap()
5434 .query(None, None, Some("<http://ex/o5>"));
5435 want.sort();
5436 let mut got = routed;
5437 got.sort();
5438 assert_eq!(got, want, "routed single-pattern read on a lean file");
5439 }
5440
5441 #[test]
5442 fn read_coalesced_merges_within_gap_and_splits_beyond() {
5443 use crate::reader::{CountingReader, SliceReader};
5444 let bytes = vec![0u8; 4096];
5445 let ranges = [
5447 ByteRange { offset: 0, len: 16 },
5448 ByteRange {
5449 offset: 48,
5450 len: 16,
5451 },
5452 ByteRange {
5453 offset: 1088,
5454 len: 16,
5455 },
5456 ];
5457 let r = CountingReader::new(SliceReader::new(&bytes));
5459 let out = read_coalesced(&r, &ranges, 16).unwrap();
5460 assert_eq!(out.len(), 3);
5461 assert_eq!(r.requests(), 3);
5462 let r = CountingReader::new(SliceReader::new(&bytes));
5464 read_coalesced(&r, &ranges, 64).unwrap();
5465 assert_eq!(r.requests(), 2);
5466 let r = CountingReader::new(SliceReader::new(&bytes));
5468 read_coalesced(&r, &ranges, 4096).unwrap();
5469 assert_eq!(r.requests(), 1);
5470 }
5471
5472 fn build_image() -> Vec<u8> {
5473 let triples = [
5474 ("Alice", "knows", "Bob"),
5475 ("Bob", "knows", "Carol"),
5476 ("Alice", "age", "30"),
5477 ];
5478 let mut db = DictionaryBuilder::new();
5479 for (s, p, o) in triples {
5480 db.observe(s, p, o);
5481 }
5482 let dict = db.build();
5483
5484 let mut ib = GraphIndexBuilder::new();
5485 for (s, p, o) in triples {
5486 ib.push(dict.encode(s, p, o).unwrap());
5487 }
5488 let index = ib.build();
5489
5490 let (meta, levels) = build_pyramid_meta(&dict, &triples_ids(&dict), DEFAULT_TILE_BUDGET);
5491 write_file(&dict, &index, false, &meta, levels)
5492 }
5493
5494 fn triples_ids(dict: &Dictionary) -> Vec<(u32, u32, u32)> {
5495 [
5496 ("Alice", "knows", "Bob"),
5497 ("Bob", "knows", "Carol"),
5498 ("Alice", "age", "30"),
5499 ]
5500 .iter()
5501 .map(|(s, p, o)| dict.encode(s, p, o).unwrap())
5502 .collect()
5503 }
5504
5505 #[test]
5506 fn file_round_trips_header_and_counts() {
5507 let bytes = build_image();
5508 let rete = Rete::open(&bytes).unwrap();
5509 assert_eq!(rete.header().quad_count, 3);
5510 assert!(rete.header().term_count >= 5);
5511 let expected_codec = writer_codec();
5512 assert_eq!(rete.header().dict_codec, expected_codec);
5513 assert_eq!(rete.header().block_codec, expected_codec);
5514 assert_eq!(&bytes[bytes.len() - 4..], &MAGIC); }
5516
5517 #[test]
5522 fn multi_tile_file_round_trips_and_routes() {
5523 let triples: Vec<(String, String, String)> = (0..200)
5524 .map(|i| {
5525 (
5526 format!("<http://ex/s/{i}>"),
5527 format!("<http://ex/p/{}>", i % 5),
5528 format!("<http://ex/o/{}>", i % 23),
5529 )
5530 })
5531 .collect();
5532 let mut db = DictionaryBuilder::new();
5533 for (s, p, o) in &triples {
5534 db.observe(s, p, o);
5535 }
5536 let dict = db.build();
5537 let mut ib = GraphIndexBuilder::new().with_tile_budget(64);
5538 for (s, p, o) in &triples {
5539 ib.push(dict.encode(s, p, o).unwrap());
5540 }
5541 let index = ib.build();
5542 assert!(
5543 index.tile_sections()[0].len() > 3,
5544 "tiny budget must force many tiles"
5545 );
5546 let bytes = write_file(&dict, &index, false, &[], 0);
5547
5548 let rete = Rete::open(&bytes).unwrap();
5549 assert_eq!(rete.header().version, crate::header::CURRENT_FORMAT_VERSION);
5550 assert_eq!(rete.query(None, None, None).len(), 200);
5551 assert_eq!(rete.query(Some("<http://ex/s/7>"), None, None).len(), 1);
5552 assert_eq!(
5553 rete.query(None, Some("<http://ex/p/3>"), None).len(),
5554 40,
5555 "predicate extent spans tiles"
5556 );
5557 assert_eq!(
5558 rete.query(None, None, Some("<http://ex/o/22>")).len(),
5559 8 );
5561
5562 use crate::reader::SliceReader;
5564 let reader = SliceReader::new(&bytes);
5565 let routed = Rete::query_ranged(&reader, Some("<http://ex/s/7>"), None, None).unwrap();
5566 assert_eq!(routed.len(), 1);
5567 let routed = Rete::query_ranged(&reader, None, Some("<http://ex/p/3>"), None).unwrap();
5568 assert_eq!(routed.len(), 40);
5569 let routed = Rete::query_ranged(&reader, None, None, Some("<http://ex/o/22>")).unwrap();
5570 assert_eq!(routed.len(), 8);
5571 }
5572
5573 #[test]
5582 fn tile_directory_offsets_survive_past_4gib() {
5583 let mut dir = Vec::new();
5584 write_uvarint(&mut dir, 2); write_uvarint(&mut dir, 5); write_uvarint(&mut dir, 0); write_uvarint(&mut dir, 3 << 30); write_uvarint(&mut dir, 1); write_uvarint(&mut dir, 0);
5590 write_uvarint(&mut dir, 2 << 30); let total = dir.len() as u64 + (3u64 << 30) + (2u64 << 30) + 64;
5592 let entries = parse_tile_directory(&dir, total).unwrap();
5593 assert_eq!(entries.len(), 2);
5594 assert_eq!(entries[1].start, dir.len() as u64 + (3u64 << 30));
5595 assert!(
5596 entries[1].end > u32::MAX as u64,
5597 "tail tile sits past 4 GiB"
5598 );
5599 assert!(parse_tile_directory(&dir, 1 << 20).is_err());
5601 }
5602
5603 #[test]
5604 fn tile_synopsis_trailer_round_trips() {
5605 let mut ib = GraphIndexBuilder::new().with_tile_budget(64);
5606 for i in 0..200u32 {
5607 ib.push((i, i % 7, i % 13));
5608 }
5609 let index = ib.build();
5610 let tile_count = index.tile_sections()[0].len();
5611 assert!(tile_count > 3, "tiny budget forces many tiles");
5612
5613 let payload = encode_tiled_section(&index, 0, CODEC_NONE);
5614 let dir = parse_tile_directory(&payload, payload.len() as u64).unwrap();
5615 assert_eq!(dir.len(), tile_count);
5616 let trailer_start = dir.iter().map(|e| e.end).max().unwrap();
5618 assert!(
5619 trailer_start < payload.len() as u64,
5620 "a trailer follows the tiles"
5621 );
5622 for e in &dir {
5623 assert!(
5624 e.end <= payload.len() as u64,
5625 "tiles still located within the payload"
5626 );
5627 }
5628 let syn = parse_tile_synopsis(&payload, trailer_start as usize, dir.len()).unwrap();
5629 for (e, (min_b, max_b, min_c, max_c)) in dir.iter().zip(syn) {
5630 let block = decompress(CODEC_NONE, &payload[e.start as usize..e.end as usize]).unwrap();
5631 let z = *crate::triples::TripleBlock::parse(&block).unwrap().zone();
5632 assert_eq!(
5633 (min_b, max_b, min_c, max_c),
5634 (z.min_b, z.max_b, z.min_c, z.max_c),
5635 "synopsis equals the tile's own zone"
5636 );
5637 }
5638 }
5639
5640 #[test]
5644 fn tile_synopsis_lazy_matches_reference_every_shape() {
5645 use crate::reader::{CountingReader, SliceReader};
5646 let triples: Vec<(String, String, String)> = (0..200u32)
5647 .map(|i| {
5648 (
5649 format!("<http://ex/s/{i:04}>"),
5650 format!("<http://ex/p/{}>", i % 7),
5651 format!("<http://ex/o/{:04}>", i % 13),
5652 )
5653 })
5654 .collect();
5655 let mut db = DictionaryBuilder::new();
5656 for (s, p, o) in &triples {
5657 db.observe(s, p, o);
5658 }
5659 let dict = db.build();
5660 let mut ib = GraphIndexBuilder::new().with_tile_budget(64);
5661 for (s, p, o) in &triples {
5662 ib.push(dict.encode(s, p, o).unwrap());
5663 }
5664 let bytes = write_file(&dict, &ib.build(), false, &[], 0);
5665
5666 let eager = Rete::open(&bytes).unwrap();
5667 assert!(
5668 eager.header().has_tile_synopsis(),
5669 "new files set the synopsis flag"
5670 );
5671
5672 let leaked: &'static [u8] = Box::leak(bytes.clone().into_boxed_slice());
5675 let reader = std::sync::Arc::new(CountingReader::new(SliceReader::new(leaked)));
5676 let lazy = Rete::open_ranged_lazy(reader).unwrap();
5677
5678 let brute = |s: Option<&str>, p: Option<&str>, o: Option<&str>| {
5679 let mut v: Vec<(String, String, String)> = triples
5680 .iter()
5681 .filter(|(a, b, c)| {
5682 s.is_none_or(|x| x == a) && p.is_none_or(|x| x == b) && o.is_none_or(|x| x == c)
5683 })
5684 .cloned()
5685 .collect();
5686 v.sort();
5687 v
5688 };
5689 let sv = [
5691 None,
5692 Some("<http://ex/s/0007>"),
5693 Some("<http://ex/s/0130>"),
5694 Some("<http://ex/s/9999>"),
5695 ];
5696 let pv = [
5697 None,
5698 Some("<http://ex/p/3>"),
5699 Some("<http://ex/p/6>"),
5700 Some("<http://ex/p/999>"),
5701 ];
5702 let ov = [
5703 None,
5704 Some("<http://ex/o/0000>"),
5705 Some("<http://ex/o/0012>"),
5706 Some("<http://ex/o/9999>"),
5707 ];
5708 for &s in &sv {
5709 for &p in &pv {
5710 for &o in &ov {
5711 let mut e = eager.query(s, p, o);
5712 e.sort();
5713 let mut l = lazy.query(s, p, o);
5714 l.sort();
5715 let r = brute(s, p, o);
5716 assert_eq!(e, r, "eager {s:?} {p:?} {o:?}");
5717 assert_eq!(l, r, "lazy {s:?} {p:?} {o:?} — synopsis over-pruned");
5718 }
5719 }
5720 }
5721 assert!(!lazy.index_incomplete(), "no lazy fetch failed");
5722 }
5723
5724 #[test]
5728 fn polyglot_offset_reads_lazily() {
5729 use crate::reader::{
5730 detect_polyglot_base, CountingReader, OffsetReader, SliceReader, POLYGLOT_DIGITS,
5731 POLYGLOT_MARKER,
5732 };
5733
5734 let image = build_image();
5738 let mut shell = Vec::new();
5739 shell.extend_from_slice(b"<!DOCTYPE html><html><head><!--");
5740 shell.extend_from_slice(POLYGLOT_MARKER);
5741 let digits_at = shell.len();
5742 shell.extend_from_slice(&[b'0'; POLYGLOT_DIGITS]); shell.extend_from_slice(b"--></head><body>a web page</body></html>\n");
5744 shell.resize(50_000, b' '); let base = shell.len() as u64;
5746 let digits = format!("{base:0width$}", width = POLYGLOT_DIGITS);
5747 shell[digits_at..digits_at + POLYGLOT_DIGITS].copy_from_slice(digits.as_bytes());
5748
5749 let mut poly = shell;
5750 poly.extend_from_slice(&image);
5751
5752 assert_ne!(
5754 &poly[0..4],
5755 b"RETE",
5756 "polyglot must not start with the magic"
5757 );
5758 let detected = detect_polyglot_base(&poly[..crate::header::HEADER_LEN]).unwrap();
5760 assert_eq!(detected, base);
5761
5762 let leaked: &'static [u8] = Box::leak(poly.into_boxed_slice());
5764 let counting = std::sync::Arc::new(CountingReader::new(SliceReader::new(leaked)));
5765 let lazy = Rete::open_ranged_lazy(OffsetReader::new(counting.clone(), detected)).unwrap();
5766
5767 let plain = Rete::open(&image).unwrap();
5769 let mut want = plain.query(None, None, None);
5770 want.sort();
5771 let mut got = lazy.query(None, None, None);
5772 got.sort();
5773 assert_eq!(
5774 got, want,
5775 "polyglot lazy query differs from the plain .rete"
5776 );
5777 assert!(!got.is_empty(), "expected some triples");
5778 assert!(!lazy.index_incomplete(), "a lazy fetch failed");
5779
5780 let read = counting.bytes_read();
5783 eprintln!(
5784 "polyglot lazy read: HTML shell {base} B + .rete image {} B; \
5785 the query touched only {read} B (never the shell).",
5786 image.len()
5787 );
5788 assert!(
5789 read < base,
5790 "read {read} bytes; the HTML shell alone is {base} — the reader \
5791 touched the prefix instead of only the embedded .rete"
5792 );
5793 }
5794
5795 #[cfg(test)]
5798 fn build_text_indexed(triples: &[(String, String, String)]) -> Vec<u8> {
5799 let mut db = DictionaryBuilder::new();
5800 for (s, p, o) in triples {
5801 db.observe(s, p, o);
5802 }
5803 let dict = db.build();
5804 let mut ib = GraphIndexBuilder::new().with_tile_budget(64);
5805 let mut id_triples: Vec<(u32, u32, u32)> = Vec::with_capacity(triples.len());
5806 for (s, p, o) in triples {
5807 let t = dict.encode(s, p, o).unwrap();
5808 ib.push(t);
5809 id_triples.push(t);
5810 }
5811 let index = ib.build();
5812 let text_index = compute_text_index(&dict, &id_triples);
5813 assert!(
5814 !text_index.is_empty(),
5815 "literals should produce a text index"
5816 );
5817 write_dataset_with_metadata(&dict, &index, &[], false, &[], 0, &[], &text_index)
5818 }
5819
5820 #[test]
5824 fn text_index_eager_matches_brute_force() {
5825 let triples: Vec<(String, String, String)> = vec![
5826 (
5827 "<http://ex/s0>",
5828 "<http://ex/label>",
5829 "\"alpha glucose phosphate\"",
5830 ),
5831 ("<http://ex/s1>", "<http://ex/label>", "\"beta Glucose\""),
5832 ("<http://ex/s2>", "<http://ex/label>", "\"gamma fructose\""),
5833 (
5834 "<http://ex/s3>",
5835 "<http://ex/note>",
5836 "\"einstein relativity\"",
5837 ),
5838 (
5839 "<http://ex/s4>",
5840 "<http://ex/ref>",
5841 "<http://ex/not-a-literal>",
5842 ),
5843 ]
5844 .into_iter()
5845 .map(|(s, p, o)| (s.to_string(), p.to_string(), o.to_string()))
5846 .collect();
5847 let bytes = build_text_indexed(&triples);
5848 let rete = Rete::open(&bytes).unwrap();
5849 assert!(rete.has_text_index());
5850
5851 let brute = |words: &[&str]| -> Vec<String> {
5853 let mut v: Vec<String> = triples
5854 .iter()
5855 .filter(|(_, _, o)| {
5856 crate::terms::is_literal(o)
5857 && words.iter().all(|w| {
5858 let wl = w.to_lowercase();
5859 crate::terms::literal_lexical(o)
5860 .unwrap()
5861 .split(|c: char| !c.is_alphanumeric())
5862 .any(|t| t.to_lowercase() == wl)
5863 })
5864 })
5865 .map(|(s, _, _)| s.clone())
5866 .collect();
5867 v.sort();
5868 v.dedup();
5869 v
5870 };
5871
5872 let mut got = rete.text_search(&["glucose"], None, 0);
5873 got.sort();
5874 assert_eq!(got, brute(&["glucose"]), "case-insensitive single word");
5875
5876 let mut got = rete.text_search(&["glucose", "phosphate"], None, 0);
5878 got.sort();
5879 assert_eq!(got, brute(&["glucose", "phosphate"]));
5880
5881 assert!(rete.text_search(&["zzznope"], None, 0).is_empty());
5883
5884 let got = rete.text_search(&[], Some("ein"), 0);
5886 assert_eq!(got, vec!["<http://ex/s3>".to_string()]);
5887
5888 let mut db = DictionaryBuilder::new();
5890 for (s, p, o) in &triples {
5891 db.observe(s, p, o);
5892 }
5893 let dict = db.build();
5894 let mut ib = GraphIndexBuilder::new();
5895 for (s, p, o) in &triples {
5896 ib.push(dict.encode(s, p, o).unwrap());
5897 }
5898 let plain = write_dataset(&dict, &ib.build(), &[], false, &[], 0);
5899 let plain_rete = Rete::open(&plain).unwrap();
5900 assert!(!plain_rete.has_text_index());
5901 assert!(plain_rete.text_search(&["glucose"], None, 0).is_empty());
5902 }
5903
5904 #[test]
5908 fn text_index_lazy_faults_only_queried_postings() {
5909 use crate::reader::{CountingReader, SliceReader};
5910 let mut triples: Vec<(String, String, String)> = (0..300u32)
5913 .map(|i| {
5914 (
5915 format!("<http://ex/s/{i:04}>"),
5916 "<http://ex/label>".to_string(),
5917 format!("\"common word number {i}\""),
5918 )
5919 })
5920 .collect();
5921 for i in [3u32, 77, 250] {
5922 triples.push((
5923 format!("<http://ex/s/{i:04}>"),
5924 "<http://ex/tag>".to_string(),
5925 "\"raretoken\"".to_string(),
5926 ));
5927 }
5928 let bytes = build_text_indexed(&triples);
5929 let eager = Rete::open(&bytes).unwrap();
5930 let mut want = eager.text_search(&["raretoken"], None, 0);
5931 want.sort();
5932 assert_eq!(want.len(), 3);
5933
5934 let leaked: &'static [u8] = Box::leak(bytes.clone().into_boxed_slice());
5935 let reader = std::sync::Arc::new(CountingReader::new(SliceReader::new(leaked)));
5936 let lazy = Rete::open_ranged_lazy(reader.clone()).unwrap();
5937 let before = reader.bytes_read();
5940 let mut got = lazy.text_search(&["raretoken"], None, 0);
5941 got.sort();
5942 assert_eq!(got, want, "lazy search matches eager");
5943 let pulled = reader.bytes_read() - before;
5944 let ti_len = eager.header().text_index_len;
5947 assert!(
5948 pulled < ti_len,
5949 "search pulled {pulled} B but the section is {ti_len} B — faulted too much"
5950 );
5951 assert!(!lazy.index_incomplete());
5952 }
5953
5954 #[test]
5959 fn search_view_matches_full_open_for_far_fewer_bytes() {
5960 use crate::reader::{CountingReader, SliceReader};
5961 let filler = "lorem ipsum dolor sit amet consectetur adipiscing elit ".repeat(20);
5964 let mut triples: Vec<(String, String, String)> = (0..300u32)
5965 .map(|i| {
5966 (
5967 format!("<http://ex/s/{i:04}>"),
5968 "<http://ex/abstract>".to_string(),
5969 format!("\"{filler} number {i}\""),
5970 )
5971 })
5972 .collect();
5973 for i in [3u32, 77, 250] {
5974 triples.push((
5975 format!("<http://ex/s/{i:04}>"),
5976 "<http://ex/tag>".to_string(),
5977 "\"raretoken\"".to_string(),
5978 ));
5979 }
5980 let bytes = build_text_indexed(&triples);
5981 let leaked: &'static [u8] = Box::leak(bytes.clone().into_boxed_slice());
5982
5983 let full_reader = std::sync::Arc::new(CountingReader::new(SliceReader::new(leaked)));
5984 let full = Rete::open_ranged_lazy(full_reader.clone()).unwrap();
5985 let mut want = full.text_search(&["raretoken"], None, 0);
5986 want.sort();
5987 assert_eq!(want.len(), 3);
5988
5989 let search_reader = std::sync::Arc::new(CountingReader::new(SliceReader::new(leaked)));
5990 let view = SearchView::open_ranged(search_reader.clone()).unwrap();
5991 assert!(view.has_text_index());
5992 let mut got = view.text_search(&["raretoken"], None, 0);
5993 got.sort();
5994 assert_eq!(got, want, "search view matches the full open");
5995 assert!(
5996 search_reader.bytes_read() < full_reader.bytes_read(),
5997 "search read {} B, full open read {} B — the narrow open bought nothing",
5998 search_reader.bytes_read(),
5999 full_reader.bytes_read()
6000 );
6001
6002 let mut prefix_hits = view.prefix_search("", 5);
6005 prefix_hits.sort();
6006 assert!(!prefix_hits.is_empty() || !view.has_pyramid());
6007 }
6008
6009 #[test]
6019 fn dict_chunk_directory_probe_costs_about_one_directory() {
6020 use crate::reader::{CountingReader, SliceReader};
6021 let head = "x".repeat(3000);
6035 let triples: Vec<(String, String, String)> = (0..2000u32)
6036 .map(|i| {
6037 let tail = char::from(b'a' + (i % 26) as u8).to_string().repeat(1000);
6038 (
6039 format!("<http://ex/s/{i:04}>"),
6040 "<http://ex/abstract>".to_string(),
6041 format!("\"{head}{i:04}{tail}\""),
6042 )
6043 })
6044 .collect();
6045 let mut db = DictionaryBuilder::new();
6046 for (s, p, o) in &triples {
6047 db.observe(s, p, o);
6048 }
6049 let dict = db.build();
6050 let mut ib = GraphIndexBuilder::new();
6051 for (s, p, o) in &triples {
6052 ib.push(dict.encode(s, p, o).unwrap());
6053 }
6054 let bytes = write_file(&dict, &ib.build(), false, &[], 0);
6055
6056 let reader = CountingReader::new(SliceReader::new(&bytes));
6057 let header = Header::from_bytes(&reader.read_at(0, HEADER_LEN as u64).unwrap()).unwrap();
6058 let section = locate_container_section_ranged(
6060 &reader,
6061 header.dictionary_offset,
6062 header.dictionary_len,
6063 2,
6064 4,
6065 )
6066 .unwrap();
6067
6068 let before = reader.bytes_read();
6069 let (_meta, entries) = read_dict_dir_ranged(&reader, section).unwrap();
6070 let spent = reader.bytes_read() - before;
6071
6072 let dir_end = entries[0].start;
6074 let head = reader.read_at(section.offset, 64).unwrap();
6075 let (header_len, n0) = read_uvarint(&head).unwrap();
6076 let dir_start = n0 as u64 + header_len;
6077 let dir_len = dir_end - dir_start;
6078 let dir_total = section.len - dir_start;
6079 assert!(
6080 dir_start + dir_len > 8192,
6081 "directory fits the header prefix ({dir_len} B) — the probe never runs"
6082 );
6083
6084 let mut old = 0u64;
6086 let mut p = 4096u64.min(dir_total).max(1);
6087 loop {
6088 let got = p.min(dir_total);
6089 old += got;
6090 if got >= dir_len || p >= dir_total {
6091 break;
6092 }
6093 p = p.saturating_mul(2).min(dir_total);
6094 }
6095 assert!(
6096 spent <= dir_len * 2,
6097 "probe read {spent} B for a {dir_len} B directory — past the append-only bound"
6098 );
6099 assert!(
6100 spent < old,
6101 "probe read {spent} B; the re-reading loop it replaced spent {old} B"
6102 );
6103 }
6104
6105 fn verbatim_keyed_twin(payload: &[u8], codec: u8) -> Vec<u8> {
6115 let (_meta, entries) = parse_chunked_dict_dir(payload, payload.len() as u64).unwrap();
6116 let (header_len, n0) = read_uvarint(payload).unwrap();
6117 let mut out = Vec::new();
6118 write_uvarint(&mut out, header_len);
6119 out.extend_from_slice(&payload[n0..n0 + header_len as usize]);
6120 write_uvarint(&mut out, entries.len() as u64);
6121 let mut prev_run = 0usize;
6122 for e in &entries {
6123 let body = decompress(codec, &payload[e.start as usize..e.end as usize]).unwrap();
6124 let first = crate::dict::run_first_term(&body, 0).unwrap();
6125 write_uvarint(&mut out, (e.first_run - prev_run) as u64);
6126 write_uvarint(&mut out, first.len() as u64);
6127 out.extend_from_slice(&first);
6128 write_uvarint(&mut out, e.end - e.start);
6129 prev_run = e.first_run;
6130 }
6131 for e in &entries {
6132 out.extend_from_slice(&payload[e.start as usize..e.end as usize]);
6133 }
6134 out
6135 }
6136
6137 fn chunk_bounds_of(payload: &[u8], codec: u8) -> Vec<(Vec<u8>, Vec<u8>)> {
6139 let (_meta, entries) = parse_chunked_dict_dir(payload, payload.len() as u64).unwrap();
6140 entries
6141 .iter()
6142 .map(|e| {
6143 let body = decompress(codec, &payload[e.start as usize..e.end as usize]).unwrap();
6144 (
6145 crate::dict::run_first_term(&body, 0).unwrap(),
6146 crate::dict::run_last_term(&body, 0, body.len()).unwrap(),
6147 )
6148 })
6149 .collect()
6150 }
6151
6152 fn adversarial_terms() -> Vec<String> {
6158 let mut terms: Vec<String> = Vec::new();
6159 for i in 0..6_000 {
6160 terms.push(format!("<http://example.org/work/{i:06}>"));
6161 }
6162 let shared = "X".repeat(900);
6163 for i in 0..4_000 {
6164 terms.push(format!("\"{shared}-{i:05}-tail\""));
6165 }
6166 let deep = "Y".repeat(3_000); for i in 0..120 {
6168 terms.push(format!("\"{deep}{i:03}\""));
6169 }
6170 for i in 0..6 {
6171 terms.push(format!("\"BIG-{i:02}-{}\"", "Z".repeat(90_000)));
6173 }
6174 for i in 0..800 {
6175 terms.push(format!(
6176 "\"h\u{e9}llo-{}-\u{1F600}{i:04}\"",
6177 "\u{e9}".repeat(40)
6178 ));
6179 }
6180 for i in 0..2_000 {
6181 terms.push(format!("\"{i:05}-{}\"", "T".repeat(1_800)));
6182 }
6183 terms.sort();
6184 terms.dedup();
6185 terms
6186 }
6187
6188 fn chunked_section_of(terms: &[String]) -> Vec<u8> {
6189 let mut b = crate::dict::DictSectionBuilder::new().with_restart_interval(16);
6190 for t in terms {
6191 b.push(t.clone());
6192 }
6193 b.build()
6194 }
6195
6196 #[test]
6206 fn chunk_directory_keys_are_separators_not_terms() {
6207 let terms = adversarial_terms();
6208 let raw = chunked_section_of(&terms);
6209 let payload = encode_chunked_dict_section(&raw, CODEC_NONE);
6210 let (_meta, entries) = parse_chunked_dict_dir(&payload, payload.len() as u64).unwrap();
6211 let bounds = chunk_bounds_of(&payload, CODEC_NONE);
6212 assert!(
6213 entries.len() > 20,
6214 "need a many-chunk section, got {}",
6215 entries.len()
6216 );
6217
6218 assert!(entries[0].key.is_empty(), "chunk 0 needs no separator");
6219 let mut key_bytes = 0usize;
6220 let mut term_bytes = 0usize;
6221 let mut differ = 0usize;
6222 for (i, e) in entries.iter().enumerate() {
6223 key_bytes += e.key.len();
6224 term_bytes += bounds[i].0.len();
6225 if e.key != bounds[i].0 {
6226 differ += 1;
6227 }
6228 assert!(
6229 e.key <= bounds[i].0,
6230 "chunk {i}: key must be <= its own first term"
6231 );
6232 if i > 0 {
6233 assert!(
6234 e.key.as_slice() > bounds[i - 1].1.as_slice(),
6235 "chunk {i}: key must be > the previous chunk's last term"
6236 );
6237 assert!(
6238 e.key > entries[i - 1].key,
6239 "chunk {i}: keys must ascend for partition_point"
6240 );
6241 assert_eq!(
6243 e.key,
6244 crate::dict::shortest_separator(&bounds[i - 1].1, &bounds[i].0),
6245 "chunk {i}: key is not the SHORTEST separator"
6246 );
6247 }
6248 }
6249 assert!(
6250 differ > 0,
6251 "every key equalled its first term — the writer stores terms again"
6252 );
6253 assert!(
6254 key_bytes * 4 < term_bytes,
6255 "separators bought almost nothing: {key_bytes} B vs {term_bytes} B of first terms"
6256 );
6257 eprintln!(
6258 "chunks={} first-term keys={term_bytes} B separator keys={key_bytes} B ({:.1}x)",
6259 entries.len(),
6260 term_bytes as f64 / key_bytes.max(1) as f64
6261 );
6262 }
6263
6264 #[test]
6275 fn separator_keys_route_identically_to_verbatim_first_terms() {
6276 let terms = adversarial_terms();
6277 let raw = chunked_section_of(&terms);
6278 let sep_payload = encode_chunked_dict_section(&raw, CODEC_NONE);
6279 let ver_payload = verbatim_keyed_twin(&sep_payload, CODEC_NONE);
6280
6281 let (_ms, es) = parse_chunked_dict_dir(&sep_payload, sep_payload.len() as u64).unwrap();
6283 let (_mv, ev) = parse_chunked_dict_dir(&ver_payload, ver_payload.len() as u64).unwrap();
6284 assert_eq!(es.len(), ev.len());
6285 for (a, b) in es.iter().zip(ev.iter()) {
6286 assert_eq!(a.first_run, b.first_run);
6287 assert_eq!(
6288 &sep_payload[a.start as usize..a.end as usize],
6289 &ver_payload[b.start as usize..b.end as usize],
6290 "chunk bodies must be identical"
6291 );
6292 }
6293 assert!(
6294 sep_payload.len() < ver_payload.len(),
6295 "separator payload is not smaller"
6296 );
6297
6298 let sec_s = decode_chunked_dict_section(
6299 &sep_payload,
6300 CODEC_NONE,
6301 crate::chunk_cache::ChunkCache::unlimited_arc(),
6302 0,
6303 )
6304 .unwrap();
6305 let sec_v = decode_chunked_dict_section(
6306 &ver_payload,
6307 CODEC_NONE,
6308 crate::chunk_cache::ChunkCache::unlimited_arc(),
6309 0,
6310 )
6311 .unwrap();
6312
6313 for (i, t) in terms.iter().enumerate() {
6315 let want = Some(i as u32 + 1);
6316 assert_eq!(sec_v.id(t), want, "verbatim id() lost a term");
6317 assert_eq!(sec_s.id(t), want, "separator id() lost a term");
6318 }
6319 for i in 0..terms.len() as u32 {
6321 assert_eq!(sec_s.term(i + 1), sec_v.term(i + 1));
6322 assert_eq!(
6323 sec_s.term(i + 1).as_deref(),
6324 Some(terms[i as usize].as_str())
6325 );
6326 }
6327 let bounds = chunk_bounds_of(&sep_payload, CODEC_NONE);
6329 let mut probes: Vec<String> = Vec::new();
6330 let push = |bytes: &[u8], probes: &mut Vec<String>| {
6331 if let Ok(s) = std::str::from_utf8(bytes) {
6332 probes.push(s.to_string());
6333 }
6334 };
6335 for (first, last) in &bounds {
6336 for base in [first, last] {
6337 push(base, &mut probes); let mut after = base.clone();
6339 after.push(0x01); push(&after, &mut probes);
6341 let mut far = base.clone();
6342 far.push(b'~');
6343 push(&far, &mut probes);
6344 if !base.is_empty() {
6345 push(&base[..base.len() - 1], &mut probes); let mut down = base.clone();
6347 *down.last_mut().unwrap() = down.last().unwrap().wrapping_sub(1);
6348 push(&down, &mut probes);
6349 let mut up = base.clone();
6350 *up.last_mut().unwrap() = up.last().unwrap().wrapping_add(1);
6351 push(&up, &mut probes);
6352 }
6353 }
6354 }
6355 for e in &es {
6357 push(&e.key, &mut probes);
6358 let mut plus = e.key.clone();
6359 plus.push(0);
6360 push(&plus, &mut probes);
6361 }
6362 probes.push(String::new()); probes.push("\u{1}".to_string()); probes.push("\u{10FFFF}\u{10FFFF}".to_string()); probes.push(terms.first().unwrap().clone());
6366 probes.push(terms.last().unwrap().clone());
6367
6368 let mut hits = 0usize;
6369 for p in &probes {
6370 let (a, b) = (sec_v.id(p), sec_s.id(p));
6371 assert_eq!(
6372 a,
6373 b,
6374 "probe diverged: {:?}",
6375 p.get(..p.len().min(48)).unwrap_or(p.as_str())
6376 );
6377 hits += usize::from(a.is_some());
6378 }
6379 assert!(
6380 probes.len() > 500 && hits > 50,
6381 "probe set is too thin: {} probes, {hits} of them present",
6382 probes.len()
6383 );
6384 eprintln!(
6385 "{} boundary probes, {hits} present, all identical",
6386 probes.len()
6387 );
6388 }
6389
6390 mod separator_props {
6396 use super::*;
6397 use proptest::prelude::*;
6398
6399 fn term_set() -> impl Strategy<Value = Vec<String>> {
6406 prop::collection::vec(("[a-c]{1,6}", 0u8..3, 300usize..1200), 60..250).prop_map(
6407 |specs| {
6408 let mut terms: Vec<String> = specs
6409 .into_iter()
6410 .enumerate()
6411 .map(|(i, (core, kind, pad))| {
6412 let block = "z".repeat(pad);
6413 match kind {
6414 0 => format!("<http://ex/{core}/{i:05}>"),
6415 1 => format!("\"{block}{core}{i:05}\""),
6416 _ => format!("\"{core}{i:05}{block}\""),
6417 }
6418 })
6419 .collect();
6420 terms.sort();
6421 terms.dedup();
6422 terms
6423 },
6424 )
6425 }
6426
6427 proptest! {
6428 #[test]
6429 fn prop_separator_keyed_section_answers_like_verbatim(
6430 terms in term_set(),
6431 extra in prop::collection::vec(any::<String>(), 0..8),
6432 ) {
6433 let raw = chunked_section_of(&terms);
6434 let sep = encode_chunked_dict_section(&raw, CODEC_NONE);
6435 let ver = verbatim_keyed_twin(&sep, CODEC_NONE);
6436 prop_assert!(sep.len() <= ver.len());
6437
6438 let (_m, es) = parse_chunked_dict_dir(&sep, sep.len() as u64).unwrap();
6440 let bounds = chunk_bounds_of(&sep, CODEC_NONE);
6441 prop_assert!(es[0].key.is_empty());
6442 for i in 1..es.len() {
6443 prop_assert!(es[i].key.as_slice() > bounds[i - 1].1.as_slice());
6444 prop_assert!(es[i].key <= bounds[i].0);
6445 prop_assert!(es[i].key > es[i - 1].key);
6446 }
6447
6448 let sec_s = decode_chunked_dict_section(
6449 &sep,
6450 CODEC_NONE,
6451 crate::chunk_cache::ChunkCache::unlimited_arc(),
6452 0,
6453 )
6454 .unwrap();
6455 let sec_v = decode_chunked_dict_section(
6456 &ver,
6457 CODEC_NONE,
6458 crate::chunk_cache::ChunkCache::unlimited_arc(),
6459 0,
6460 )
6461 .unwrap();
6462
6463 for (i, t) in terms.iter().enumerate() {
6464 prop_assert_eq!(sec_s.id(t), Some(i as u32 + 1));
6465 prop_assert_eq!(sec_v.id(t), Some(i as u32 + 1));
6466 prop_assert_eq!(sec_s.term(i as u32 + 1), sec_v.term(i as u32 + 1));
6467 }
6468
6469 let mut probes: Vec<String> = extra;
6471 probes.push(String::new());
6472 for (first, last) in &bounds {
6473 for base in [first, last] {
6474 if let Ok(s) = std::str::from_utf8(base) {
6475 probes.push(s.to_string());
6476 probes.push(format!("{s}\u{1}"));
6477 probes.push(s[..s.len() - s.chars().next_back()
6478 .map(char::len_utf8).unwrap_or(0)].to_string());
6479 }
6480 }
6481 }
6482 for e in &es {
6483 if let Ok(s) = std::str::from_utf8(&e.key) {
6484 probes.push(s.to_string());
6485 }
6486 }
6487 for p in &probes {
6488 prop_assert_eq!(sec_s.id(p), sec_v.id(p), "probe diverged: {:?}", p);
6489 }
6490 }
6491 }
6492 }
6493
6494 #[test]
6499 fn text_index_is_tamper_evident_and_verifies() {
6500 let triples: Vec<(String, String, String)> = vec![(
6501 "<http://ex/s0>".to_string(),
6502 "<http://ex/label>".to_string(),
6503 "\"alpha glucose phosphate\"".to_string(),
6504 )];
6505 let bytes = build_text_indexed(&triples);
6506 let header = Rete::open(&bytes).unwrap().header().clone();
6507 assert!(header.text_index_len > 0);
6508 assert!(verify(&bytes).unwrap(), "a text-indexed build must verify");
6509
6510 let mut tampered = bytes.clone();
6511 tampered[header.text_index_offset as usize] ^= 0xff;
6512 assert!(
6513 !verify(&tampered).unwrap(),
6514 "tampering with the text index must break verify()"
6515 );
6516 }
6517
6518 #[test]
6522 fn synopsis_cuts_remote_fetch_bytes() {
6523 use crate::header::FLAG_TILE_SYNOPSIS;
6524 use crate::reader::{CountingReader, SliceReader};
6525
6526 let triples: Vec<(String, String, String)> = (0..400u32)
6530 .map(|i| {
6531 (
6532 format!("<http://ex/s/{i:04}>"),
6533 "<http://ex/p>".to_string(),
6534 format!("<http://ex/o/{i:04}>"),
6535 )
6536 })
6537 .collect();
6538 let mut db = DictionaryBuilder::new();
6539 for (s, p, o) in &triples {
6540 db.observe(s, p, o);
6541 }
6542 let dict = db.build();
6543 let mut ib = GraphIndexBuilder::new().with_tile_budget(64);
6544 for (s, p, o) in &triples {
6545 ib.push(dict.encode(s, p, o).unwrap());
6546 }
6547 let bytes = write_file(&dict, &ib.build(), false, &[], 0);
6548
6549 let q = (Some("<http://ex/s/0395>"), None, Some("<http://ex/o/0005>"));
6553 let query_bytes = |image: &[u8]| -> (u64, usize) {
6557 let leaked: &'static [u8] = Box::leak(image.to_vec().into_boxed_slice());
6558 let reader = std::sync::Arc::new(CountingReader::new(SliceReader::new(leaked)));
6559 let rete = Rete::open_ranged_lazy(reader.clone()).unwrap();
6560 let before = reader.bytes_read(); let n = rete.query(q.0, q.1, q.2).len();
6562 assert!(!rete.index_incomplete());
6563 (reader.bytes_read() - before, n)
6564 };
6565
6566 let (on_bytes, on_n) = query_bytes(&bytes);
6567 let mut off = bytes.clone();
6569 off[5] &= !FLAG_TILE_SYNOPSIS;
6570 let (off_bytes, off_n) = query_bytes(&off);
6571
6572 assert_eq!(on_n, 0, "the pair never co-occurs");
6573 assert_eq!(off_n, 0, "same answer without the synopsis");
6574 assert!(
6578 on_bytes < off_bytes,
6579 "synopsis skips the routed tile fetch: {on_bytes} < {off_bytes}"
6580 );
6581 }
6582
6583 #[test]
6589 fn double_bound_object_join_eager_matches_lazy() {
6590 use crate::reader::SliceReader;
6591 let occ = "<http://ex/occ>";
6592 let phys = "<http://ex/physicist>";
6593 let phil = "<http://ex/philosopher>";
6594 let label = "<http://www.w3.org/2000/01/rdf-schema#label>";
6595 let mut triples: Vec<(String, String, String)> = Vec::new();
6597 for i in 0..20u32 {
6598 triples.push((format!("<http://ex/p/{i:02}>"), occ.into(), phys.into()));
6599 if i < 10 {
6600 triples.push((format!("<http://ex/p/{i:02}>"), occ.into(), phil.into()));
6601 }
6602 triples.push((
6603 format!("<http://ex/p/{i:02}>"),
6604 label.into(),
6605 format!("\"Name {i:02}\""),
6606 ));
6607 }
6608 let mut db = DictionaryBuilder::new();
6609 for (s, p, o) in &triples {
6610 db.observe(s, p, o);
6611 }
6612 let dict = db.build();
6613 let mut ib = GraphIndexBuilder::new().with_tile_budget(16);
6614 for (s, p, o) in &triples {
6615 ib.push(dict.encode(s, p, o).unwrap());
6616 }
6617 let bytes = write_file(&dict, &ib.build(), false, &[], 0);
6618
6619 let q = "SELECT ?l WHERE { \
6620 ?p <http://ex/occ> <http://ex/physicist> ; \
6621 <http://ex/occ> <http://ex/philosopher> ; \
6622 <http://www.w3.org/2000/01/rdf-schema#label> ?l }";
6623 let run = |rete: &Rete| -> Vec<String> {
6624 let (_, sols) = crate::eval_sparql(rete, q).unwrap();
6625 let mut v: Vec<String> = sols.iter().filter_map(|b| b.get("l").cloned()).collect();
6626 v.sort();
6627 v
6628 };
6629
6630 let eager_rows = run(&Rete::open(&bytes).unwrap());
6631 let leaked: &'static [u8] = Box::leak(bytes.clone().into_boxed_slice());
6632 let lazy = Rete::open_ranged_lazy(std::sync::Arc::new(SliceReader::new(leaked))).unwrap();
6633 let lazy_rows = run(&lazy);
6634 assert!(!lazy.index_incomplete());
6635
6636 assert_eq!(eager_rows.len(), 10, "the 10 physicist∩philosopher labels");
6637 assert_eq!(eager_rows, lazy_rows, "eager and lazy must agree exactly");
6638 }
6639
6640 #[test]
6644 fn multi_chunk_dictionary_round_trips() {
6645 let mut db = DictionaryBuilder::new();
6646 let term = |i: u32| format!("<http://example.org/some/long/prefix/entity/{i:06}>");
6647 for i in 0..6000u32 {
6648 db.observe(&term(i), "<http://ex/p>", &term(i + 1));
6649 }
6650 let dict = db.build();
6651 let mut ib = GraphIndexBuilder::new();
6652 for i in 0..6000u32 {
6653 ib.push(
6654 dict.encode(&term(i), "<http://ex/p>", &term(i + 1))
6655 .unwrap(),
6656 );
6657 }
6658 let bytes = write_file(&dict, &ib.build(), false, &[], 0);
6659 let rete = Rete::open(&bytes).unwrap();
6660 let d = rete.dictionary();
6661 assert_eq!(d.term_count(), dict.term_count());
6662 for i in (0..6000).step_by(97).chain([0, 1, 5999, 6000]) {
6663 let t = term(i);
6664 let sid = dict.subject_id(&t);
6665 assert_eq!(d.subject_id(&t), sid, "subject_id({t})");
6666 if let Some(id) = sid {
6667 assert_eq!(d.subject_term(id).as_deref(), Some(t.as_str()));
6668 }
6669 let oid = dict.object_id(&t);
6670 assert_eq!(d.object_id(&t), oid, "object_id({t})");
6671 }
6672 assert_eq!(d.subject_id("<http://example.org/absent>"), None);
6673 assert_eq!(d.predicate_id("<http://ex/p>"), Some(1));
6674 assert_eq!(d.predicate_term(1).as_deref(), Some("<http://ex/p>"));
6675 }
6676
6677 #[test]
6678 #[cfg(feature = "compression")]
6679 fn compression_shrinks_repetitive_data() {
6680 let mut db = DictionaryBuilder::new();
6684 let triples: Vec<(String, String, String)> = (0..500)
6685 .map(|i| {
6686 (
6687 format!("<http://example.org/entity/{i}>"),
6688 "<http://example.org/p/relatedTo>".to_string(),
6689 format!("<http://example.org/entity/{}>", (i + 1) % 500),
6690 )
6691 })
6692 .collect();
6693 for (s, p, o) in &triples {
6694 db.observe(s, p, o);
6695 }
6696 let dict = db.build();
6697 let mut ib = GraphIndexBuilder::new();
6698 for (s, p, o) in &triples {
6699 ib.push(dict.encode(s, p, o).unwrap());
6700 }
6701 let bytes = write_file(&dict, &ib.build(), false, &[], 0);
6702
6703 let raw: usize = triples
6704 .iter()
6705 .map(|(s, p, o)| s.len() + p.len() + o.len())
6706 .sum();
6707 assert!(
6708 bytes.len() < raw / 2,
6709 "expected strong compression: file {} vs raw terms {raw}",
6710 bytes.len()
6711 );
6712
6713 let rete = Rete::open(&bytes).unwrap();
6715 let r = rete.query(Some("<http://example.org/entity/0>"), None, None);
6716 assert_eq!(r.len(), 1);
6717 assert_eq!(r[0].2, "<http://example.org/entity/1>");
6718 }
6719
6720 fn big_file_with_pyramid() -> Vec<u8> {
6721 let triples: Vec<(String, String, String)> = (0..300)
6723 .map(|i| {
6724 (
6725 format!("<http://ex/e{i}>"),
6726 "<http://ex/next>".to_string(),
6727 format!("<http://ex/e{}>", (i + 1) % 300),
6728 )
6729 })
6730 .collect();
6731 let mut db = DictionaryBuilder::new();
6732 for (s, p, o) in &triples {
6733 db.observe(s, p, o);
6734 }
6735 let dict = db.build();
6736 let ids: Vec<_> = triples
6737 .iter()
6738 .map(|(s, p, o)| dict.encode(s, p, o).unwrap())
6739 .collect();
6740 let mut ib = GraphIndexBuilder::new();
6741 for &t in &ids {
6742 ib.push(t);
6743 }
6744 let (meta, levels) = build_pyramid_meta(&dict, &ids, DEFAULT_TILE_BUDGET);
6745 write_file(&dict, &ib.build(), false, &meta, levels)
6746 }
6747
6748 #[test]
6749 fn ranged_open_is_minimal_and_correct() {
6750 use crate::reader::{CountingReader, SliceReader};
6751 let bytes = big_file_with_pyramid();
6752
6753 let full = CountingReader::new(SliceReader::new(&bytes));
6754 let rete = Rete::open_ranged(&full).unwrap();
6755 assert!(full.requests() <= 4, "requests = {}", full.requests());
6757 assert_eq!(
6758 rete.query(Some("<http://ex/e0>"), None, None)[0].2,
6759 "<http://ex/e1>"
6760 );
6761
6762 let summ_reader = CountingReader::new(SliceReader::new(&bytes));
6764 let view = SummaryView::open_ranged(&summ_reader).unwrap().unwrap();
6765 assert!(!view.summary.is_empty());
6766 assert!(
6767 summ_reader.bytes_read() < bytes.len() as u64,
6768 "summary read {} of {} bytes",
6769 summ_reader.bytes_read(),
6770 bytes.len()
6771 );
6772 assert!(summ_reader.bytes_read() < full.bytes_read());
6774 }
6775
6776 #[test]
6777 fn content_hash_is_set_and_verifies() {
6778 let bytes = build_image();
6779 let rete = Rete::open(&bytes).unwrap();
6780 assert_ne!(
6781 rete.header().content_hash,
6782 [0u8; 16],
6783 "hash must be populated"
6784 );
6785 assert!(verify(&bytes).unwrap(), "freshly built file verifies");
6786 assert!(
6787 verify_ranged(&crate::reader::SliceReader::new(&bytes)).unwrap(),
6788 "the streaming verify agrees"
6789 );
6790
6791 assert_eq!(
6793 Rete::open(&build_image()).unwrap().header().content_hash,
6794 rete.header().content_hash
6795 );
6796
6797 let mut tampered = bytes.clone();
6799 let last = tampered.len() - 5; tampered[last] ^= 0xff;
6801 assert!(!verify(&tampered).unwrap());
6802 assert!(!verify_ranged(&crate::reader::SliceReader::new(&tampered)).unwrap());
6803 let short = &bytes[..bytes.len() / 2];
6805 assert!(verify_ranged(&crate::reader::SliceReader::new(short)).is_err());
6806 assert!(verify_ranged(&crate::reader::SliceReader::new(&bytes[..10])).is_err());
6807 }
6808
6809 fn build_with_metadata(meta: &[u8]) -> Vec<u8> {
6811 let triples = [
6812 ("Alice", "knows", "Bob"),
6813 ("Bob", "knows", "Carol"),
6814 ("Alice", "age", "30"),
6815 ];
6816 let mut db = DictionaryBuilder::new();
6817 for (s, p, o) in triples {
6818 db.observe(s, p, o);
6819 }
6820 let dict = db.build();
6821 let ids: Vec<_> = triples
6822 .iter()
6823 .map(|(s, p, o)| dict.encode(s, p, o).unwrap())
6824 .collect();
6825 let mut ib = GraphIndexBuilder::new();
6826 for &t in &ids {
6827 ib.push(t);
6828 }
6829 let (pmeta, levels) = build_pyramid_meta(&dict, &ids, DEFAULT_TILE_BUDGET);
6830 write_dataset_with_metadata(&dict, &ib.build(), &[], false, &pmeta, levels, meta, &[])
6831 }
6832
6833 #[test]
6834 fn metadata_round_trips_and_shifts_offsets() {
6835 let card = br#"{"title":"My Dataset"}"#;
6836 let bytes = build_with_metadata(card);
6837 let rete = Rete::open(&bytes).unwrap();
6838
6839 assert_eq!(rete.metadata(), Some(card.as_slice()));
6841 let h = rete.header();
6842 assert_eq!(h.metadata_offset, HEADER_LEN as u64);
6843 assert_eq!(h.metadata_len, card.len() as u64);
6844 assert_eq!(h.dictionary_offset, HEADER_LEN as u64 + card.len() as u64);
6846
6847 assert_eq!(
6849 rete.query(Some("Bob"), Some("knows"), Some("Carol")).len(),
6850 1
6851 );
6852 assert!(verify(&bytes).unwrap());
6854 }
6855
6856 #[test]
6857 fn empty_metadata_is_byte_identical_to_plain_writer() {
6858 assert_eq!(
6861 build_with_metadata(&[]),
6862 build_image(),
6863 "empty-metadata output must equal the plain writer byte-for-byte"
6864 );
6865 }
6866
6867 #[test]
6872 fn replace_metadata_matches_a_direct_write() {
6873 let first = br#"{"title":"My Dataset","queries":[1,2,3]}"#;
6874 let bytes = build_with_metadata(first);
6875 for target in [
6876 br#"{"title":"My Dataset","queries":[1,2]}"#.as_slice(), br#"{"title":"My Dataset","queries":[1,2,3,4,5,6,7]}"#.as_slice(), b"".as_slice(), first.as_slice(), ] {
6881 let spliced = replace_metadata(&bytes, target).unwrap();
6882 assert_eq!(
6883 spliced,
6884 build_with_metadata(target),
6885 "splicing {} bytes of metadata must equal a direct write",
6886 target.len()
6887 );
6888 assert!(
6889 verify(&spliced).unwrap(),
6890 "the new hash covers the new card"
6891 );
6892 let rete = Rete::open(&spliced).unwrap();
6893 assert_eq!(
6894 rete.metadata(),
6895 (!target.is_empty()).then_some(target),
6896 "the new payload reads back verbatim"
6897 );
6898 assert_eq!(
6900 rete.query(Some("Bob"), Some("knows"), Some("Carol")).len(),
6901 1
6902 );
6903 }
6904 }
6905
6906 #[test]
6909 fn replace_metadata_keeps_build_info_adjacent() {
6910 let with_info =
6911 attach_build_info(&build_with_metadata(br#"{"a":1}"#), br#"{"builder":"x"}"#).unwrap();
6912 let spliced = replace_metadata(&with_info, br#"{"a":1,"b":2}"#).unwrap();
6913 let h = Header::from_bytes(&spliced).unwrap();
6914 assert_eq!(h.metadata_offset, HEADER_LEN as u64);
6915 assert_eq!(h.build_info_offset, HEADER_LEN as u64 + h.metadata_len);
6916 assert_eq!(
6917 read_build_info(&spliced).unwrap().as_deref(),
6918 Some(br#"{"builder":"x"}"#.as_slice())
6919 );
6920 assert!(verify(&spliced).unwrap());
6921 assert_eq!(
6923 h.content_hash,
6924 Header::from_bytes(&build_with_metadata(br#"{"a":1,"b":2}"#))
6925 .unwrap()
6926 .content_hash
6927 );
6928 }
6929
6930 #[test]
6931 fn metadata_is_tamper_evident() {
6932 let card = br#"{"title":"x"}"#;
6933 let mut bytes = build_with_metadata(card);
6934 assert!(verify(&bytes).unwrap());
6935 bytes[HEADER_LEN + 2] ^= 0xff;
6937 assert!(
6938 !verify(&bytes).unwrap(),
6939 "tampering with the card must break verify()"
6940 );
6941 }
6942
6943 #[test]
6944 fn ranged_opens_do_not_fetch_metadata() {
6945 use crate::reader::{CountingReader, SliceReader};
6946 let card = vec![0xABu8; 512]; let bytes = build_with_metadata(&card);
6948 let total = bytes.len() as u64;
6949
6950 let r = CountingReader::new(SliceReader::new(&bytes));
6952 let rete = Rete::open_ranged(&r).unwrap();
6953 assert!(
6954 rete.metadata().is_none(),
6955 "open_ranged must not load the card"
6956 );
6957 assert!(r.requests() <= 4, "requests = {}", r.requests());
6958 assert!(
6959 r.bytes_read() <= total - card.len() as u64,
6960 "read {} of {} bytes; the {}-byte card must be skipped",
6961 r.bytes_read(),
6962 total,
6963 card.len()
6964 );
6965
6966 let rs = CountingReader::new(SliceReader::new(&bytes));
6968 let view = SummaryView::open_ranged(&rs).unwrap().unwrap();
6969 assert!(!view.summary.is_empty());
6970 assert!(rs.bytes_read() <= total - card.len() as u64);
6971 }
6972
6973 #[test]
6974 fn metadata_ranged_fetches_only_header_and_card() {
6975 use crate::reader::{CountingReader, SliceReader};
6976 let card = vec![0xCDu8; 384];
6979 let bytes = build_with_metadata(&card);
6980
6981 let r = CountingReader::new(SliceReader::new(&bytes));
6982 let got = read_metadata_ranged(&r).unwrap().unwrap();
6983 assert_eq!(got, card, "the card reads back verbatim");
6984 assert_eq!(r.requests(), 2, "exactly header + metadata ranges");
6985 assert_eq!(
6986 r.bytes_read(),
6987 HEADER_LEN as u64 + card.len() as u64,
6988 "no dictionary/index/pyramid bytes are touched"
6989 );
6990
6991 let plain = build_image();
6993 let rp = CountingReader::new(SliceReader::new(&plain));
6994 assert!(read_metadata_ranged(&rp).unwrap().is_none());
6995 assert_eq!(rp.requests(), 1, "header only for a cardless file");
6996 assert_eq!(rp.bytes_read(), HEADER_LEN as u64);
6997 }
6998
6999 #[test]
7000 fn build_info_attaches_outside_the_hash() {
7001 let card = br#"{"title":"My Dataset"}"#;
7002 let base = build_with_metadata(card);
7003 let base_hash = Header::from_bytes(&base).unwrap().content_hash;
7004
7005 let info = br#"{"built_at":"2026-08-04T00:00:00Z","builder":"rete-cli 0.3.2"}"#;
7006 let with = attach_build_info(&base, info).unwrap();
7007
7008 assert_eq!(read_build_info(&with).unwrap().unwrap(), info);
7010 let h = Header::from_bytes(&with).unwrap();
7011 assert_eq!(h.build_info_offset, HEADER_LEN as u64 + card.len() as u64);
7012 assert_eq!(h.build_info_len, info.len() as u64);
7013 assert_eq!(h.expected_file_len(), Some(with.len() as u64));
7014
7015 assert_eq!(h.content_hash, base_hash);
7019 assert!(verify(&with).unwrap());
7020 let other = attach_build_info(&base, br#"{"built_at":"1999-01-01T00:00:00Z"}"#).unwrap();
7021 assert_eq!(Header::from_bytes(&other).unwrap().content_hash, base_hash);
7022
7023 let rete = Rete::open(&with).unwrap();
7025 assert_eq!(
7026 rete.query(Some("Bob"), Some("knows"), Some("Carol")).len(),
7027 1
7028 );
7029 assert_eq!(rete.metadata(), Some(card.as_slice()));
7030
7031 let replaced = attach_build_info(&with, br#"{"builder":"other"}"#).unwrap();
7034 assert_eq!(
7035 read_build_info(&replaced).unwrap().unwrap(),
7036 br#"{"builder":"other"}"#
7037 );
7038 assert!(verify(&replaced).unwrap());
7039 let stripped = attach_build_info(&replaced, &[]).unwrap();
7040 assert_eq!(stripped, base, "stripping build-info restores the image");
7041 }
7042
7043 #[test]
7044 fn build_info_attaches_on_a_metadata_free_file() {
7045 let base = build_image();
7047 let info = b"{\"builder\":\"x\"}";
7048 let with = attach_build_info(&base, info).unwrap();
7049 let h = Header::from_bytes(&with).unwrap();
7050 assert_eq!(h.build_info_offset, HEADER_LEN as u64);
7051 assert_eq!(h.metadata_len, 0);
7052 assert_eq!(read_build_info(&with).unwrap().unwrap(), info);
7053 assert!(verify(&with).unwrap());
7054 assert_eq!(
7055 h.content_hash,
7056 Header::from_bytes(&base).unwrap().content_hash
7057 );
7058 Rete::open(&with).unwrap();
7059 }
7060
7061 #[test]
7068 fn a_plan_rebuilds_exactly_what_the_in_memory_splice_produces() {
7069 let card = br#"{"title":"My Dataset"}"#;
7070 let base = build_with_metadata(card);
7071 let base_hash = Header::from_bytes(&base).unwrap().content_hash;
7072 for start in [
7073 base.clone(),
7074 attach_build_info(&base, b"{\"a\":1}").unwrap(),
7075 ] {
7076 for info in [
7077 br#"{"builder":"rete-cli 0.3.2","query_costs":{}}"#.as_slice(), b"{}".as_slice(), b"".as_slice(), ] {
7081 let plan = plan_build_info(&start, start.len() as u64, info.len() as u64).unwrap();
7082 let mut streamed = Vec::new();
7083 streamed.extend_from_slice(&plan.header);
7084 streamed.extend_from_slice(&start[HEADER_LEN..plan.insert as usize]);
7085 streamed.extend_from_slice(info);
7086 streamed.extend_from_slice(&start[plan.tail_start as usize..]);
7087 assert_eq!(streamed.len() as u64, plan.new_len);
7088 assert_eq!(
7089 streamed,
7090 attach_build_info(&start, info).unwrap(),
7091 "a {}-byte section streamed must equal the same section spliced",
7092 info.len()
7093 );
7094 assert!(verify(&streamed).unwrap());
7096 let h = Header::from_bytes(&streamed).unwrap();
7097 assert_eq!(h.content_hash, base_hash);
7098 assert_eq!(h.expected_file_len(), Some(plan.new_len));
7099 let rete = Rete::open(&streamed).unwrap();
7100 assert_eq!(rete.metadata(), Some(card.as_slice()));
7101 assert_eq!(
7102 rete.query(Some("Bob"), Some("knows"), Some("Carol")).len(),
7103 1
7104 );
7105 }
7106 }
7107 }
7108
7109 #[test]
7112 fn a_plan_refuses_a_header_that_overruns_the_file() {
7113 let base = build_with_metadata(br#"{"title":"x"}"#);
7114 assert!(plan_build_info(&base, 8, 16).is_err());
7115 assert!(plan_build_info(&base[..HEADER_LEN], base.len() as u64, 16).is_ok());
7116 }
7117
7118 #[test]
7119 fn card_and_build_info_ranged_is_one_header_plus_one_range() {
7120 use crate::reader::{CountingReader, SliceReader};
7121 let card = vec![0xCDu8; 384];
7122 let info = vec![0xEFu8; 200];
7123 let bytes = attach_build_info(&build_with_metadata(&card), &info).unwrap();
7124
7125 let r = CountingReader::new(SliceReader::new(&bytes));
7128 let (m, b) = read_card_and_build_info_ranged(&r).unwrap();
7129 assert_eq!(m.unwrap(), card);
7130 assert_eq!(b.unwrap(), info);
7131 assert_eq!(r.requests(), 2, "header + one coalesced range");
7132 assert_eq!(
7133 r.bytes_read(),
7134 HEADER_LEN as u64 + card.len() as u64 + info.len() as u64
7135 );
7136
7137 let plain = build_with_metadata(&card);
7139 let rp = CountingReader::new(SliceReader::new(&plain));
7140 let (m, b) = read_card_and_build_info_ranged(&rp).unwrap();
7141 assert_eq!(m.unwrap(), card);
7142 assert!(b.is_none());
7143 assert_eq!(rp.requests(), 2);
7144 assert_eq!(
7149 r.requests(),
7150 rp.requests(),
7151 "build info must not cost an extra request",
7152 );
7153
7154 let none = build_image();
7156 let rn = CountingReader::new(SliceReader::new(&none));
7157 let (m, b) = read_card_and_build_info_ranged(&rn).unwrap();
7158 assert!(m.is_none() && b.is_none());
7159 assert_eq!(rn.requests(), 1);
7160 }
7161
7162 #[test]
7163 fn schema_summary_groups_by_type() {
7164 let rt = RDF_TYPE;
7165 let bytes = build_from(&[
7166 ("Alice", rt, "Person"),
7167 ("Bob", rt, "Person"),
7168 ("NYC", rt, "City"),
7169 ("Alice", "knows", "Bob"),
7170 ("Alice", "livesIn", "NYC"),
7171 ("Alice", "name", "\"Alice\""),
7172 ]);
7173 let rete = Rete::open(&bytes).unwrap();
7174 let summary = schema_summary(&rete);
7175 assert!(summary.contains(&("Person".into(), "knows".into(), "Person".into(), 1)));
7177 assert!(summary.contains(&("Person".into(), "livesIn".into(), "City".into(), 1)));
7178 assert!(summary.contains(&("Person".into(), "name".into(), "(literal)".into(), 1)));
7179 assert!(!summary.iter().any(|(_, p, _, _)| p == RDF_TYPE));
7181
7182 let classes = schema_classes(&rete);
7184 assert_eq!(
7185 classes,
7186 vec![("Person".into(), 2u32), ("City".into(), 1u32)]
7187 );
7188 }
7189
7190 fn build_from(triples: &[(&str, &str, &str)]) -> Vec<u8> {
7191 let mut db = DictionaryBuilder::new();
7192 for (s, p, o) in triples {
7193 db.observe(s, p, o);
7194 }
7195 let dict = db.build();
7196 let mut ib = GraphIndexBuilder::new();
7197 for (s, p, o) in triples {
7198 ib.push(dict.encode(s, p, o).unwrap());
7199 }
7200 write_file(&dict, &ib.build(), false, &[], 0)
7201 }
7202
7203 fn build_with_pyramid(triples: &[(&str, &str, &str)]) -> Vec<u8> {
7205 let mut db = DictionaryBuilder::new();
7206 for (s, p, o) in triples {
7207 db.observe(s, p, o);
7208 }
7209 let dict = db.build();
7210 let encoded: Vec<_> = triples
7211 .iter()
7212 .map(|(s, p, o)| dict.encode(s, p, o).unwrap())
7213 .collect();
7214 let mut ib = GraphIndexBuilder::new();
7215 for t in &encoded {
7216 ib.push(*t);
7217 }
7218 let (meta, levels) = build_pyramid_meta(&dict, &encoded, DEFAULT_TILE_BUDGET);
7219 write_dataset(&dict, &ib.build(), &[], false, &meta, levels)
7220 }
7221
7222 #[test]
7223 fn tbox_coherence_flags_unsatisfiable_class_index_free() {
7224 use crate::reader::{CountingReader, SliceReader};
7225 let rt = RDF_TYPE;
7226 let sub = "<http://www.w3.org/2000/01/rdf-schema#subClassOf>";
7227 let disj = "<http://www.w3.org/2002/07/owl#disjointWith>";
7228 let bytes = build_with_pyramid(&[
7231 ("<http://ex/C>", sub, "<http://ex/D>"),
7232 ("<http://ex/C>", sub, "<http://ex/E>"),
7233 ("<http://ex/D>", disj, "<http://ex/E>"),
7234 ("<http://ex/x>", rt, "<http://ex/C>"),
7235 ]);
7236
7237 let r = CountingReader::new(SliceReader::new(&bytes));
7238 let view = SummaryView::open_ranged(&r).unwrap().unwrap();
7239 let points = view.tbox_coherence();
7240 assert!(
7241 points
7242 .iter()
7243 .any(|i| i.kind == "unsatisfiable-class" && i.detail.contains("http://ex/C>")),
7244 "expected C unsatisfiable from the schema alone, got {points:?}"
7245 );
7246
7247 let header = Header::from_bytes(&bytes[..HEADER_LEN]).unwrap();
7250 assert!(
7251 r.bytes_read() <= bytes.len() as u64 - header.root_dir_len,
7252 "tbox_coherence must not read the triple index"
7253 );
7254 }
7255
7256 #[test]
7257 fn schema_coherence_reads_only_the_schema_block() {
7258 use crate::reader::{CountingReader, SliceReader};
7259 let rt = RDF_TYPE;
7260 let sub = "<http://www.w3.org/2000/01/rdf-schema#subClassOf>";
7261 let disj = "<http://www.w3.org/2002/07/owl#disjointWith>";
7262 let mut triples: Vec<(String, String, String)> = vec![
7266 ("<http://ex/C>".into(), sub.into(), "<http://ex/D>".into()),
7267 ("<http://ex/C>".into(), sub.into(), "<http://ex/E>".into()),
7268 ("<http://ex/D>".into(), disj.into(), "<http://ex/E>".into()),
7269 ];
7270 for i in 0..500 {
7271 let s = format!("<http://ex/x{i}>");
7272 triples.push((s.clone(), rt.into(), "<http://ex/C>".into()));
7273 triples.push((
7274 s,
7275 "<http://ex/label>".into(),
7276 format!("\"unique label {i}\""),
7277 ));
7278 }
7279 let trefs: Vec<(&str, &str, &str)> = triples
7280 .iter()
7281 .map(|(s, p, o)| (s.as_str(), p.as_str(), o.as_str()))
7282 .collect();
7283 let bytes = build_with_pyramid(&trefs);
7284
7285 let header = Header::from_bytes(&bytes[..HEADER_LEN]).unwrap();
7286 assert!(
7287 header.schema_meta_len > 0,
7288 "the writer recorded a schema-block length"
7289 );
7290 assert!(
7291 (header.schema_meta_len as u64) < header.pyramid_meta_len,
7292 "schema block ({}) should be far smaller than the whole pyramid-meta ({})",
7293 header.schema_meta_len,
7294 header.pyramid_meta_len
7295 );
7296
7297 let r = CountingReader::new(SliceReader::new(&bytes));
7298 let points = read_schema_coherence_ranged(&r).unwrap().unwrap();
7299 assert!(points.iter().any(|i| i.kind == "unsatisfiable-class"));
7300 assert!(
7302 r.bytes_read() <= HEADER_LEN as u64 + header.schema_meta_len as u64,
7303 "read {} bytes; expected <= header + schema block ({})",
7304 r.bytes_read(),
7305 HEADER_LEN as u64 + header.schema_meta_len as u64
7306 );
7307 }
7308
7309 #[test]
7310 fn tbox_coherence_clean_schema_is_coherent() {
7311 use crate::reader::SliceReader;
7312 let rt = RDF_TYPE;
7313 let sub = "<http://www.w3.org/2000/01/rdf-schema#subClassOf>";
7314 let bytes = build_with_pyramid(&[
7315 ("<http://ex/Dog>", sub, "<http://ex/Animal>"),
7316 ("<http://ex/x>", rt, "<http://ex/Dog>"),
7317 ]);
7318 let view = SummaryView::open_ranged(&SliceReader::new(&bytes))
7319 .unwrap()
7320 .unwrap();
7321 assert!(view.tbox_is_coherent(), "a plain hierarchy is coherent");
7322 }
7323
7324 #[test]
7325 fn named_graphs_round_trip() {
7326 let all = [
7328 ("Alice", "knows", "Bob"), ("Bob", "age", "30"), ];
7331 let mut db = DictionaryBuilder::new();
7332 for (s, p, o) in all {
7333 db.observe(s, p, o);
7334 }
7335 let dict = db.build();
7336
7337 let mut def = GraphIndexBuilder::new();
7338 def.push(dict.encode("Alice", "knows", "Bob").unwrap());
7339 let mut g1 = GraphIndexBuilder::new();
7340 g1.push(dict.encode("Bob", "age", "30").unwrap());
7341
7342 let named = vec![("http://ex/g1".to_string(), g1.build())];
7343 let bytes = write_dataset(&dict, &def.build(), &named, true, &[], 0);
7344
7345 assert!(verify(&bytes).unwrap());
7346 let rete = Rete::open(&bytes).unwrap();
7347 assert_eq!(rete.graph_names(), vec!["http://ex/g1"]);
7348 let gi = rete.graph_index("http://ex/g1").unwrap();
7350 assert_eq!(gi.triple_count(), 1);
7351 assert!(rete.graph_index("http://ex/missing").is_none());
7352
7353 assert_eq!(rete.header().quad_count, 2);
7356
7357 assert_eq!(rete.query(Some("Alice"), None, None).len(), 1);
7359
7360 assert_eq!(
7362 rete.dump(None),
7363 vec![("Alice".into(), "knows".into(), "Bob".into())]
7364 );
7365 assert_eq!(
7366 rete.dump(Some("http://ex/g1")),
7367 vec![("Bob".into(), "age".into(), "30".into())]
7368 );
7369 }
7370
7371 #[test]
7377 fn named_graph_container_opens_tile_lazily_like_the_root() {
7378 let mut db = DictionaryBuilder::new();
7379 let node = |n: u32| format!("<http://ex/n{n}>");
7380 let p = "<http://ex/p>".to_string();
7381 db.observe(&node(0), &p, &node(1));
7382 for i in 0..500u32 {
7383 db.observe(&node(1000 + i), &p, &node(2000 + i));
7384 }
7385 let dict = db.build();
7386 let mut def = GraphIndexBuilder::new();
7387 def.push(dict.encode(&node(0), &p, &node(1)).unwrap());
7388 let mut g = GraphIndexBuilder::new().with_tile_budget(256); for i in 0..500u32 {
7390 g.push(dict.encode(&node(1000 + i), &p, &node(2000 + i)).unwrap());
7391 }
7392 let named = vec![("<http://ex/g>".to_string(), g.build())];
7393 let bytes = write_dataset(&dict, &def.build(), &named, true, &[], 0);
7394 let rete = Rete::open(&bytes).unwrap();
7395 let header = rete.header();
7396
7397 let soff = header.named_graphs_offset as usize;
7400 let send = soff + header.named_graphs_len as usize;
7401 let sec = &bytes[soff..send];
7402 let (n, mut pos) = read_uvarint(sec).unwrap();
7403 assert_eq!(n, 1);
7404 let (ilen, u1) = read_uvarint(&sec[pos..]).unwrap();
7405 pos += u1 + ilen as usize;
7406 let (clen, u2) = read_uvarint(&sec[pos..]).unwrap();
7407 pos += u2;
7408 let container = ByteRange {
7409 offset: (soff + pos) as u64,
7410 len: clen,
7411 };
7412
7413 use crate::reader::SliceReader;
7414 let leaked: &'static [u8] = Box::leak(bytes.clone().into_boxed_slice());
7415 let reader: std::sync::Arc<dyn RangeReader + Send + Sync> =
7416 std::sync::Arc::new(SliceReader::new(leaked));
7417 let (lazy_idx, _, _) = open_index_container_lazy(
7418 &reader,
7419 container,
7420 header.block_codec,
7421 header.has_tile_synopsis(),
7422 1,
7423 header.perms,
7424 )
7425 .unwrap();
7426 let mut want = rete
7427 .graph_index("<http://ex/g>")
7428 .unwrap()
7429 .match_pattern((None, None, None));
7430 let mut got = lazy_idx.match_pattern((None, None, None));
7431 want.sort_unstable();
7432 got.sort_unstable();
7433 assert_eq!(got.len(), 500);
7434 assert_eq!(got, want);
7435 assert!(!lazy_idx.load_incomplete());
7436 }
7437
7438 #[test]
7444 fn dump_iter_matches_dump_and_stops_early() {
7445 use crate::reader::{CountingReader, SliceReader};
7446
7447 let triples: Vec<(String, String, String)> = (0..2000u32)
7449 .map(|i| {
7450 (
7451 format!("<http://ex/s/{i:04}>"),
7452 "<http://ex/p>".to_string(),
7453 format!("<http://ex/o/{i:04}>"),
7454 )
7455 })
7456 .collect();
7457 let mut db = DictionaryBuilder::new();
7458 for (s, p, o) in &triples {
7459 db.observe(s, p, o);
7460 }
7461 let dict = db.build();
7462 let mut ib = GraphIndexBuilder::new().with_tile_budget(256);
7463 for (s, p, o) in &triples {
7464 ib.push(dict.encode(s, p, o).unwrap());
7465 }
7466 let bytes = write_file(&dict, &ib.build(), false, &[], 0);
7467
7468 let rete = Rete::open(&bytes).unwrap();
7470 assert_eq!(rete.dump_iter(None).collect::<Vec<_>>(), rete.dump(None));
7471 assert_eq!(rete.dump_iter(Some("http://ex/nope")).count(), 0);
7473
7474 let read_bytes = |take: usize| -> u64 {
7476 let leaked: &'static [u8] = Box::leak(bytes.clone().into_boxed_slice());
7477 let reader = std::sync::Arc::new(CountingReader::new(SliceReader::new(leaked)));
7478 let rete = Rete::open_ranged_lazy(reader.clone()).unwrap();
7479 let before = reader.bytes_read();
7480 let n = rete.dump_iter(None).take(take).count();
7481 assert_eq!(n, take.min(triples.len()));
7482 assert!(!rete.index_incomplete());
7483 reader.bytes_read() - before
7484 };
7485 let first_only = read_bytes(1);
7486 let everything = read_bytes(triples.len());
7487 assert!(
7489 first_only < everything,
7490 "dump_iter is not lazy: one triple read {first_only} B, a full scan {everything} B"
7491 );
7492 #[cfg(feature = "compression")]
7499 assert!(
7500 first_only * 2 < everything,
7501 "dump_iter is not lazy: one triple read {first_only} B of the {everything} B a full scan reads"
7502 );
7503
7504 let fresh = Rete::open(&bytes).unwrap();
7509 let drain = |max_quads: usize| {
7514 let (mut out, mut cursor, mut calls) = (Vec::new(), 0u32, 0);
7515 loop {
7516 let (batch, next, done) = fresh.dump_batch(None, cursor, max_quads);
7517 out.extend(batch);
7518 cursor = next;
7519 calls += 1;
7520 assert!(calls < 10_000, "cursor failed to advance");
7521 if done {
7522 break out;
7523 }
7524 }
7525 };
7526 let eager = fresh.dump(None);
7527 for max in [1usize, 7, 128, 100_000] {
7528 assert_eq!(drain(max), eager, "batch size {max} changed the dump");
7529 }
7530 let (t, _, done) = fresh.dump_batch(Some("http://ex/nope"), 0, 16);
7533 assert!(t.is_empty() && done);
7534 }
7535
7536 #[test]
7544 fn query_batch_matches_query_in_graph_and_stops_early() {
7545 use crate::reader::{CountingReader, SliceReader};
7546
7547 let triples: Vec<(String, String, String)> = (0..2000u32)
7550 .map(|i| {
7551 (
7552 format!("<http://ex/s/{:04}>", i / 2),
7553 format!("<http://ex/p/{}>", i % 2),
7554 format!("<http://ex/o/{i:04}>"),
7555 )
7556 })
7557 .collect();
7558 let mut db = DictionaryBuilder::new();
7559 for (s, p, o) in &triples {
7560 db.observe(s, p, o);
7561 }
7562 let dict = db.build();
7563 let mut ib = GraphIndexBuilder::new().with_tile_budget(256);
7564 for (s, p, o) in &triples {
7565 ib.push(dict.encode(s, p, o).unwrap());
7566 }
7567 let bytes = write_file(&dict, &ib.build(), false, &[], 0);
7568 let rete = Rete::open(&bytes).unwrap();
7569
7570 let drain = |s: Option<&str>, p: Option<&str>, o: Option<&str>, max: usize| {
7571 let (mut out, mut cursor, mut calls) = (Vec::new(), 0u64, 0);
7572 loop {
7573 let (batch, next, done) = rete.query_batch(None, s, p, o, cursor, max);
7574 assert!(!batch.is_empty() || done, "empty non-final batch");
7577 out.extend(batch);
7578 cursor = next;
7579 calls += 1;
7580 assert!(calls < 20_000, "cursor failed to advance");
7581 if done {
7582 break out;
7583 }
7584 }
7585 };
7586 let shapes: [(Option<&str>, Option<&str>, Option<&str>); 6] = [
7587 (None, None, None),
7588 (Some("<http://ex/s/0007>"), None, None),
7589 (None, Some("<http://ex/p/1>"), None),
7590 (None, None, Some("<http://ex/o/0042>")),
7591 (Some("<http://ex/s/0007>"), Some("<http://ex/p/0>"), None),
7592 (
7593 Some("<http://ex/s/0007>"),
7594 Some("<http://ex/p/0>"),
7595 Some("<http://ex/o/0014>"),
7596 ),
7597 ];
7598 for (s, p, o) in shapes {
7599 let eager = rete.query_in_graph(None, s, p, o);
7600 assert!(
7601 !eager.is_empty(),
7602 "fixture shape {s:?} {p:?} {o:?} matches nothing"
7603 );
7604 assert_eq!(
7606 rete.query_iter(None, s, p, o).collect::<Vec<_>>(),
7607 eager,
7608 "query_iter disagrees for {s:?} {p:?} {o:?}"
7609 );
7610 for max in [1usize, 3, 64, 100_000] {
7611 let mut got = drain(s, p, o, max);
7612 let mut want = eager.clone();
7613 got.sort();
7616 want.sort();
7617 assert_eq!(got, want, "batch size {max} changed {s:?} {p:?} {o:?}");
7618 }
7619 if s.is_none() && p.is_none() && o.is_none() {
7621 assert_eq!(drain(s, p, o, 7), eager, "unbound batching reordered rows");
7622 }
7623 }
7624
7625 let (t, _, done) = rete.query_batch(None, Some("<http://ex/nope>"), None, None, 0, 16);
7628 assert!(t.is_empty() && done);
7629 let (t, _, done) = rete.query_batch(Some("http://ex/nope"), None, None, None, 0, 16);
7630 assert!(t.is_empty() && done);
7631 assert_eq!(
7632 rete.query_iter(None, Some("<http://ex/nope>"), None, None)
7633 .count(),
7634 0
7635 );
7636
7637 let read_bytes = |take: usize| -> u64 {
7640 let leaked: &'static [u8] = Box::leak(bytes.clone().into_boxed_slice());
7641 let reader = std::sync::Arc::new(CountingReader::new(SliceReader::new(leaked)));
7642 let rete = Rete::open_ranged_lazy(reader.clone()).unwrap();
7643 let before = reader.bytes_read();
7644 let (mut n, mut cursor) = (0usize, 0u64);
7645 loop {
7646 let (batch, next, done) = rete.query_batch(None, None, None, None, cursor, 1);
7647 n += batch.len();
7648 cursor = next;
7649 if done || n >= take {
7650 break;
7651 }
7652 }
7653 assert!(n >= take.min(triples.len()));
7654 assert!(!rete.index_incomplete());
7655 reader.bytes_read() - before
7656 };
7657 let first_only = read_bytes(1);
7658 let everything = read_bytes(triples.len());
7659 assert!(
7660 first_only < everything,
7661 "query_batch is not lazy: one row read {first_only} B, a full scan {everything} B"
7662 );
7663 #[cfg(feature = "compression")]
7664 assert!(
7665 first_only * 2 < everything,
7666 "query_batch is not lazy: one row read {first_only} B of the {everything} B a full scan reads"
7667 );
7668 }
7669
7670 #[test]
7673 fn query_batch_is_graph_scoped() {
7674 let mut db = DictionaryBuilder::new();
7675 for (s, p, o) in [
7676 ("Alice", "knows", "Bob"),
7677 ("Alice", "knows", "Carol"),
7678 ("Alice", "knows", "Dave"),
7679 ("Erin", "knows", "Frank"),
7680 ("Alice", "knows", "Zoe"),
7681 ] {
7682 db.observe(s, p, o);
7683 }
7684 let dict = db.build();
7685 let mut def = GraphIndexBuilder::new();
7686 def.push(dict.encode("Alice", "knows", "Bob").unwrap());
7687 def.push(dict.encode("Alice", "knows", "Carol").unwrap());
7688 let mut g1 = GraphIndexBuilder::new();
7689 g1.push(dict.encode("Alice", "knows", "Dave").unwrap());
7690 g1.push(dict.encode("Erin", "knows", "Frank").unwrap());
7691 let mut g2 = GraphIndexBuilder::new();
7692 g2.push(dict.encode("Alice", "knows", "Zoe").unwrap());
7693 let named = vec![
7694 ("http://ex/g1".to_string(), g1.build()),
7695 ("http://ex/g2".to_string(), g2.build()),
7696 ];
7697 let bytes = write_dataset(&dict, &def.build(), &named, true, &[], 0);
7698 let rete = Rete::open(&bytes).unwrap();
7699 for graph in [None, Some("http://ex/g1"), Some("http://ex/g2")] {
7700 let eager = rete.query_in_graph(graph, None, None, None);
7701 let (mut out, mut cursor) = (Vec::new(), 0u64);
7702 loop {
7703 let (batch, next, done) = rete.query_batch(graph, None, None, None, cursor, 1);
7704 out.extend(batch);
7705 cursor = next;
7706 if done {
7707 break;
7708 }
7709 }
7710 assert_eq!(out, eager, "graph {graph:?} batched differently");
7711 }
7712 }
7713
7714 #[test]
7715 fn query_in_graph_is_graph_scoped() {
7716 let mut db = DictionaryBuilder::new();
7719 for (s, p, o) in [
7720 ("Alice", "knows", "Bob"),
7721 ("Alice", "knows", "Carol"),
7722 ("Alice", "knows", "Dave"),
7723 ] {
7724 db.observe(s, p, o);
7725 }
7726 let dict = db.build();
7727
7728 let mut def = GraphIndexBuilder::new();
7729 def.push(dict.encode("Alice", "knows", "Bob").unwrap());
7730 def.push(dict.encode("Alice", "knows", "Carol").unwrap());
7731 let mut g1 = GraphIndexBuilder::new();
7732 g1.push(dict.encode("Alice", "knows", "Dave").unwrap());
7733
7734 let named = vec![("http://ex/g1".to_string(), g1.build())];
7735 let bytes = write_dataset(&dict, &def.build(), &named, true, &[], 0);
7736 let rete = Rete::open(&bytes).unwrap();
7737
7738 let mut def_objs: Vec<String> = rete
7740 .query_in_graph(None, Some("Alice"), Some("knows"), None)
7741 .into_iter()
7742 .map(|(_, _, o)| o)
7743 .collect();
7744 def_objs.sort();
7745 assert_eq!(def_objs, vec!["Bob".to_string(), "Carol".to_string()]);
7746
7747 assert_eq!(
7749 rete.query_in_graph(Some("http://ex/g1"), Some("Alice"), None, None),
7750 vec![("Alice".into(), "knows".into(), "Dave".into())]
7751 );
7752
7753 assert_eq!(rete.query_in_graph(None, None, None, None).len(), 2);
7755 assert_eq!(
7756 rete.query_in_graph(Some("http://ex/g1"), None, None, None)
7757 .len(),
7758 1
7759 );
7760
7761 assert!(rete
7763 .query_in_graph(Some("http://ex/missing"), None, None, None)
7764 .is_empty());
7765 }
7766
7767 #[test]
7780 fn bounded_query_in_graph_does_not_fault_the_whole_dictionary() {
7781 use crate::reader::{CountingReader, SliceReader};
7782
7783 let payload = |i: u32| -> String {
7789 let mut s = String::with_capacity(800);
7790 let mut x = i as u64 ^ 0x9e37_79b9_7f4a_7c15;
7791 while s.len() < 800 {
7792 x = x
7793 .wrapping_mul(6364136223846793005)
7794 .wrapping_add(1442695040888963407);
7795 s.push_str(&format!("{:016x}", x ^ (x >> 29)));
7796 }
7797 s
7798 };
7799 let quads: Vec<(String, String, String, bool)> = (0..2000u32)
7806 .map(|i| {
7807 let named = i % 2 == 0;
7808 let ns = if named { 'a' } else { 'b' };
7809 (
7810 format!("<http://ex/{ns}/s/{i:04}>"),
7811 "<http://ex/abstract>".to_string(),
7812 format!("\"{ns}{} number {i}\"", payload(i)),
7813 named,
7814 )
7815 })
7816 .collect();
7817 let mut db = DictionaryBuilder::new();
7818 for (s, p, o, _) in &quads {
7819 db.observe(s, p, o);
7820 }
7821 let dict = db.build();
7822 let mut def = GraphIndexBuilder::new().with_tile_budget(256);
7823 let mut g1 = GraphIndexBuilder::new().with_tile_budget(256);
7824 for (s, p, o, named) in &quads {
7825 let t = dict.encode(s, p, o).unwrap();
7826 if *named {
7827 g1.push(t);
7828 } else {
7829 def.push(t);
7830 }
7831 }
7832 let named = vec![("http://ex/g1".to_string(), g1.build())];
7833 let bytes = write_dataset(&dict, &def.build(), &named, true, &[], 0);
7834
7835 let eager = Rete::open(&bytes).unwrap();
7836 let dict_len = eager.header().dictionary_len;
7837
7838 let leaked: &'static [u8] = Box::leak(bytes.clone().into_boxed_slice());
7841 let probe = |f: &dyn Fn(&Rete) -> Vec<TermTriple>| -> (Vec<TermTriple>, u64) {
7842 let reader = std::sync::Arc::new(CountingReader::new(SliceReader::new(leaked)));
7843 let lazy = Rete::open_ranged_lazy(reader.clone()).unwrap();
7844 let before = reader.bytes_read();
7845 let out = f(&lazy);
7846 assert!(!lazy.index_incomplete(), "a chunk fetch failed");
7847 (out, reader.bytes_read() - before)
7848 };
7849
7850 let want = eager.query_in_graph(
7852 Some("http://ex/g1"),
7853 Some("<http://ex/a/s/0100>"),
7854 None,
7855 None,
7856 );
7857 assert_eq!(want.len(), 1, "fixture: expected exactly one match");
7858 let (got, pulled) = probe(&|r| {
7859 r.query_in_graph(
7860 Some("http://ex/g1"),
7861 Some("<http://ex/a/s/0100>"),
7862 None,
7863 None,
7864 )
7865 });
7866 assert_eq!(got, want, "lazy query_in_graph disagrees with eager");
7867 assert!(
7868 pulled * 4 < dict_len,
7869 "a 1-row query_in_graph faulted {pulled} B of a {dict_len} B dictionary — \
7870 it is prefetching the whole thing again"
7871 );
7872
7873 let (empty, q_pulled) =
7881 probe(&|r| r.query_in_graph(Some("http://ex/nope"), None, None, None));
7882 assert!(empty.is_empty());
7883 let (_, d_pulled) = probe(&|r| {
7884 let mut n = 0usize;
7885 r.dump_each(Some("http://ex/nope"), |_, _, _| n += 1);
7886 assert_eq!(n, 0, "an absent graph yielded triples");
7887 Vec::new()
7888 });
7889 assert_eq!(
7890 q_pulled, d_pulled,
7891 "query_in_graph and dump_each disagree on what an absent graph costs"
7892 );
7893 assert!(
7894 q_pulled * 4 < dict_len,
7895 "an absent graph faulted {q_pulled} B of a {dict_len} B dictionary"
7896 );
7897
7898 let want: Vec<TermTriple> = eager.dump(Some("http://ex/g1"));
7901 assert_eq!(want.len(), 1000);
7902 let (got, pulled) = probe(&|r| {
7903 let mut out = Vec::new();
7904 r.dump_each(Some("http://ex/g1"), |s, p, o| {
7905 out.push((s.to_string(), p.to_string(), o.to_string()))
7906 });
7907 out
7908 });
7909 assert_eq!(got, want, "lazy dump_each disagrees with eager dump");
7910 assert!(
7911 pulled * 3 < dict_len * 2,
7912 "dumping half the graph faulted {pulled} B of a {dict_len} B dictionary — \
7913 the other graph's chunks are being pulled in too"
7914 );
7915 }
7916
7917 #[cfg(test)]
7922 fn build_multichunk_image() -> Vec<u8> {
7923 use crate::index::PermSet;
7924 let mut db = DictionaryBuilder::new();
7925 let mut triples: Vec<(String, String, String)> = (0..4000u32)
7926 .map(|i| {
7927 (
7928 format!("<http://ex/s/{i:07}>"),
7929 format!("<http://ex/p/{}>", i % 8),
7930 format!("\"object literal number {i:07} with padding text to widen the term\""),
7931 )
7932 })
7933 .collect();
7934 triples.push((
7935 "<http://ex/s/big>".to_string(),
7936 "<http://ex/p/big>".to_string(),
7937 format!("\"{}\"", "Z".repeat(8000)),
7938 ));
7939 for (s, p, o) in &triples {
7940 db.observe(s, p, o);
7941 }
7942 let dict = db.build();
7943 let mut idx = GraphIndexBuilder::new()
7944 .with_tile_budget(64)
7945 .with_perms(PermSet::ALL);
7946 for (s, p, o) in &triples {
7947 idx.push(dict.encode(s, p, o).unwrap());
7948 }
7949 write_dataset(&dict, &idx.build(), &[], true, &[], 0)
7950 }
7951
7952 #[test]
7961 fn tiny_dict_cache_evicts_yet_dump_is_byte_identical() {
7962 use crate::reader::SliceReader;
7963 let leaked: &'static [u8] = Box::leak(build_multichunk_image().into_boxed_slice());
7964 let dump = |cap: u64| -> (
7965 Vec<(String, String, String)>,
7966 crate::chunk_cache::CacheStats,
7967 ) {
7968 let rete = Rete::open_ranged_lazy(SliceReader::new(leaked)).unwrap();
7969 rete.dict.set_cache_cap(cap);
7970 let mut out = Vec::new();
7971 rete.dump_filtered_each(None, None, None, None, |s, p, o| {
7972 out.push((s.to_string(), p.to_string(), o.to_string()))
7973 });
7974 assert!(
7975 !rete.index_incomplete(),
7976 "no lazy fault should fail at cap {cap}"
7977 );
7978 let stats = rete.dict_cache_stats();
7979 (out, stats)
7980 };
7981
7982 let (unlimited, us) = dump(u64::MAX);
7983 assert_eq!(unlimited.len(), 4001, "every triple dumped");
7984 assert_eq!(us.evictions, 0, "unlimited cap evicts nothing");
7985
7986 let (tiny, ts) = dump(128 * 1024);
7989 assert_eq!(tiny, unlimited, "128 KiB-cap dump differs from unlimited");
7990 assert!(ts.evictions > 0, "a tiny cap must evict (stats {ts:?})");
7991
7992 let (starved, ss) = dump(1);
7994 assert_eq!(starved, unlimited, "1-byte-cap dump differs from unlimited");
7995 assert!(
7996 ss.evictions > 0,
7997 "1-byte cap evicts every insert (stats {ss:?})"
7998 );
7999 }
8000
8001 #[test]
8012 fn memory_budget_charges_the_fixed_directories() {
8013 use crate::reader::SliceReader;
8014 let leaked: &'static [u8] = Box::leak(build_multichunk_image().into_boxed_slice());
8015 let rete = Rete::open_ranged_lazy(SliceReader::new(leaked)).unwrap();
8016
8017 let fixed = rete.fixed_overhead();
8018 assert!(
8021 fixed.dict_directory > 0 && fixed.index_directory > 0,
8022 "{fixed:?}"
8023 );
8024 assert_eq!(fixed.named_graphs, 0, "no named graphs in this file");
8025 assert_eq!(
8026 fixed.total(),
8027 fixed.dict_directory + fixed.index_directory + fixed.named_graphs
8028 );
8029
8030 let total: u64 = 512 << 20;
8033 let applied = rete.set_memory_budget(Some(total));
8034 let plain = split_memory_budget(total);
8035 assert_eq!(applied.dict_cache, plain.dict_cache - fixed.total());
8036 assert_eq!(rete.dict_cache_cap(), applied.dict_cache);
8037 assert_eq!(applied.tile_cache, plain.tile_cache);
8038 assert_eq!(applied.block_cache, plain.block_cache);
8039 assert!(
8040 applied.dict_cache + applied.tile_cache + applied.block_cache + fixed.total() <= total,
8041 "caps + fixed directories exceed the budget: {applied:?} + {fixed:?} > {total}"
8042 );
8043
8044 let applied = rete.set_memory_budget(Some(1 << 20));
8046 assert_eq!(applied.dict_cache, MIN_DICT_CACHE_CAP);
8047
8048 let applied = rete.set_memory_budget(None);
8050 assert_eq!(applied.dict_cache, u64::MAX);
8051 assert_eq!(rete.dict_cache_cap(), u64::MAX);
8052
8053 let rete = Rete::open_ranged_lazy(SliceReader::new(leaked)).unwrap();
8056 rete.dict.set_cache_cap(128 * 1024);
8057 let mut rows = 0usize;
8058 rete.dump_filtered_each(None, None, None, None, |_, _, _| rows += 1);
8059 assert_eq!(rows, 4001);
8060 let s = rete.dict_cache_stats();
8061 assert!(s.evictions > 0, "the cap must bind: {s:?}");
8062 assert!(
8065 s.entries <= 1 || s.resident_bytes <= 128 * 1024,
8066 "dictionary cache over its cap after the dump: {s:?}"
8067 );
8068 }
8069
8070 #[test]
8075 fn releasing_a_named_graph_drops_and_reopens_its_index() {
8076 use crate::index::PermSet;
8077 use crate::reader::SliceReader;
8078 let mut db = DictionaryBuilder::new();
8079 let all: Vec<(String, String, String)> = (0..40u32)
8080 .map(|i| {
8081 (
8082 format!("<http://ex/s{}>", i % 7),
8083 format!("<http://ex/p{}>", i % 3),
8084 format!("<http://ex/o{}>", i % 5),
8085 )
8086 })
8087 .collect();
8088 for (s, p, o) in &all {
8089 db.observe(s, p, o);
8090 }
8091 let dict = db.build();
8092 let mk = |take: fn(u32) -> bool| {
8093 let mut g = GraphIndexBuilder::new()
8094 .with_tile_budget(16)
8095 .with_perms(PermSet::ALL);
8096 for (i, (s, p, o)) in all.iter().enumerate() {
8097 if take(i as u32) {
8098 g.push(dict.encode(s, p, o).unwrap());
8099 }
8100 }
8101 g.build()
8102 };
8103 let def = mk(|_| true);
8104 let g1 = mk(|i| i % 2 == 0);
8105 let g2 = mk(|i| i % 3 == 0);
8106 let image = write_dataset(
8107 &dict,
8108 &def,
8109 &[
8110 ("<http://ex/g1>".to_string(), g1),
8111 ("<http://ex/g2>".to_string(), g2),
8112 ],
8113 true,
8114 &[],
8115 0,
8116 );
8117 let leaked: &'static [u8] = Box::leak(image.into_boxed_slice());
8118 let mut rete = Rete::open_ranged_lazy(SliceReader::new(leaked)).unwrap();
8119
8120 let opened = |r: &Rete| -> usize {
8121 match &r.named_graphs {
8122 NamedGraphsSlot::Lazy(l) => {
8123 let mut n = 0;
8124 l.for_each_opened(|_| n += 1);
8125 n
8126 }
8127 _ => 0,
8128 }
8129 };
8130
8131 assert_eq!(opened(&rete), 0, "nothing opened before first touch");
8132 assert!(rete.graph_index("<http://ex/g1>").is_some());
8134 assert_eq!(opened(&rete), 1, "g1 opened");
8135 rete.release_named_graph("<http://ex/g1>");
8137 assert_eq!(opened(&rete), 0, "g1 released");
8138 rete.release_named_graph("<http://ex/absent>");
8140 rete.release_named_graph("<http://ex/g1>");
8141 assert_eq!(opened(&rete), 0);
8142 assert!(rete.graph_index("<http://ex/g1>").is_some());
8144 assert_eq!(opened(&rete), 1, "g1 re-opened after release");
8145 }
8146
8147 #[test]
8165 fn filtered_dump_is_exactly_the_unfiltered_dump_filtered() {
8166 use crate::index::PermSet;
8167 use crate::reader::SliceReader;
8168
8169 let terms = |i: u32| {
8170 (
8171 format!("<http://ex/s{}>", i % 11),
8172 format!("<http://ex/p{}>", i % 3),
8173 if i.is_multiple_of(4) {
8177 format!("\"lit {}\"@en", i % 5)
8178 } else {
8179 format!("<http://ex/o{}>", i % 7)
8180 },
8181 )
8182 };
8183 let build = |perms: PermSet| {
8184 let mut db = DictionaryBuilder::new();
8185 let all: Vec<_> = (0..90u32).map(terms).collect();
8186 for (s, p, o) in &all {
8187 db.observe(s, p, o);
8188 }
8189 let dict = db.build();
8190 let mut def = GraphIndexBuilder::new()
8191 .with_tile_budget(16)
8192 .with_perms(perms);
8193 let mut g1 = GraphIndexBuilder::new()
8194 .with_tile_budget(16)
8195 .with_perms(perms);
8196 for (i, (s, p, o)) in all.iter().enumerate() {
8197 let t = dict.encode(s, p, o).unwrap();
8198 if i % 2 == 0 {
8202 g1.push(t);
8203 }
8204 def.push(t);
8205 }
8206 write_dataset(
8207 &dict,
8208 &def.build(),
8209 &[("<http://ex/g1>".to_string(), g1.build())],
8210 true,
8211 &[],
8212 0,
8213 )
8214 };
8215
8216 let shapes: Vec<(Option<&str>, Option<&str>, Option<&str>)> = vec![
8217 (None, None, None),
8218 (Some("<http://ex/s3>"), None, None),
8219 (None, Some("<http://ex/p1>"), None),
8220 (None, None, Some("<http://ex/o5>")),
8221 (None, None, Some("\"lit 0\"@en")),
8222 (Some("<http://ex/s3>"), Some("<http://ex/p0>"), None),
8223 (Some("<http://ex/s4>"), None, Some("<http://ex/o4>")),
8224 (None, Some("<http://ex/p1>"), Some("<http://ex/o5>")),
8225 (
8226 Some("<http://ex/s3>"),
8227 Some("<http://ex/p0>"),
8228 Some("<http://ex/o3>"),
8229 ),
8230 (Some("<http://ex/nope>"), None, None),
8234 (None, Some("<http://ex/nope>"), None),
8235 (None, None, Some("<http://ex/nope>")),
8236 ];
8237
8238 let mut any_pruned = false;
8239 for perms in [PermSet::ALL, PermSet::CORE] {
8240 let image = build(perms);
8241 let leaked: &'static [u8] = Box::leak(image.clone().into_boxed_slice());
8242 let opens: Vec<(&str, Rete)> = vec![
8243 ("resident", Rete::open(&image).unwrap()),
8244 (
8245 "ranged",
8246 Rete::open_ranged(&SliceReader::new(leaked)).unwrap(),
8247 ),
8248 (
8249 "lazy",
8250 Rete::open_ranged_lazy(SliceReader::new(leaked)).unwrap(),
8251 ),
8252 ];
8253 for graph in [None, Some("<http://ex/g1>"), Some("<http://ex/absent>")] {
8254 let full: Vec<TermTriple> = {
8256 let mut v = Vec::new();
8257 Rete::open(&image).unwrap().dump_each(graph, |s, p, o| {
8258 v.push((s.to_string(), p.to_string(), o.to_string()))
8259 });
8260 v.sort();
8261 v
8262 };
8263 for &(s, p, o) in &shapes {
8264 let mut want = full.clone();
8265 want.retain(|(ts, tp, to)| {
8266 s.is_none_or(|x| x == ts)
8267 && p.is_none_or(|x| x == tp)
8268 && o.is_none_or(|x| x == to)
8269 });
8270 for (path, rete) in &opens {
8271 let mut got = Vec::new();
8272 rete.dump_filtered_each(graph, s, p, o, |s, p, o| {
8273 got.push((s.to_string(), p.to_string(), o.to_string()))
8274 });
8275 got.sort();
8276 assert_eq!(
8277 got,
8278 want,
8279 "{path}/{perms:?} graph={graph:?} pattern={:?} — filtered dump \
8280 is not the unfiltered dump filtered",
8281 (s, p, o)
8282 );
8283 assert!(!rete.index_incomplete(), "{path}: a fetch failed");
8284 }
8285 let plan = opens[2].1.dump_plan(graph, s, p, o);
8289 if graph == Some("<http://ex/absent>")
8290 || [s, p, o].contains(&Some("<http://ex/nope>"))
8291 {
8292 assert!(
8293 plan.scan.is_none(),
8294 "an unmatchable dump still planned a scan: {:?}",
8295 (graph, s, p, o)
8296 );
8297 } else {
8298 let scan = plan.scan.expect("a matchable dump plans a scan");
8299 assert!(
8300 PermSet::CORE.contains(scan.permutation),
8301 "a dump routed to {:?}, outside PermSet::CORE — a \
8302 --permutations 3 file has no such section",
8303 scan.permutation
8304 );
8305 assert!(scan.tiles_admitted <= scan.tiles_routed);
8306 assert!(scan.tiles_routed <= scan.tiles_total);
8307 assert!(scan.tile_bytes <= scan.section_bytes);
8308 if scan.tiles_admitted < scan.tiles_total {
8309 any_pruned = true;
8310 }
8311 }
8312 }
8313 }
8314 }
8315 assert!(
8316 any_pruned,
8317 "no shape pruned a single tile — the fixture is too small to be \
8318 measuring anything"
8319 );
8320 }
8321
8322 #[test]
8332 fn a_predicate_scoped_dump_fetches_less_than_the_graph() {
8333 use crate::reader::{CountingReader, SliceReader};
8334
8335 let mut db = DictionaryBuilder::new();
8341 let mut all = Vec::new();
8342 for s in 0..200u32 {
8343 for p in 0..20u32 {
8344 let t = (
8345 format!("<http://ex/s/{s:04}>"),
8346 format!("<http://ex/p/{p:02}>"),
8347 format!("\"object {s:04}/{p:02}\""),
8348 );
8349 db.observe(&t.0, &t.1, &t.2);
8350 all.push(t);
8351 }
8352 }
8353 let dict = db.build();
8354 let mut ib = GraphIndexBuilder::new().with_tile_budget(64);
8355 for (s, p, o) in &all {
8356 ib.push(dict.encode(s, p, o).unwrap());
8357 }
8358 let image = write_dataset(&dict, &ib.build(), &[], false, &[], 0);
8359 let leaked: &'static [u8] = Box::leak(image.clone().into_boxed_slice());
8360
8361 let probe = |s: Option<&str>, p: Option<&str>, o: Option<&str>| -> (usize, u64) {
8363 let reader = std::sync::Arc::new(CountingReader::new(SliceReader::new(leaked)));
8364 let lazy = Rete::open_ranged_lazy(reader.clone()).unwrap();
8365 let before = reader.bytes_read();
8366 let mut n = 0usize;
8367 lazy.dump_filtered_each(None, s, p, o, |_, _, _| n += 1);
8368 assert!(!lazy.index_incomplete());
8369 (n, reader.bytes_read() - before)
8370 };
8371
8372 let (full_rows, full_bytes) = probe(None, None, None);
8373 let (pred_rows, pred_bytes) = probe(None, Some("<http://ex/p/07>"), None);
8374 let (subj_rows, subj_bytes) = probe(Some("<http://ex/s/0021>"), None, None);
8375 assert_eq!(full_rows, 4000);
8376 assert_eq!(pred_rows, 200, "one predicate covers every subject once");
8377 assert_eq!(subj_rows, 20);
8378 assert!(
8380 pred_bytes < full_bytes && subj_bytes < full_bytes,
8381 "a slice fetched {pred_bytes} / {subj_bytes} B where the whole graph fetched \
8382 {full_bytes} B — the scan is not being pruned at all"
8383 );
8384 #[cfg(feature = "compression")]
8395 assert!(
8396 pred_bytes * 4 < full_bytes && subj_bytes * 4 < full_bytes,
8397 "a 1-in-20 predicate slice fetched {pred_bytes} B and one subject \
8398 {subj_bytes} B where the whole graph fetched {full_bytes} B"
8399 );
8400
8401 let reader = std::sync::Arc::new(CountingReader::new(SliceReader::new(leaked)));
8403 let lazy = Rete::open_ranged_lazy(reader.clone()).unwrap();
8404 let before = reader.bytes_read();
8405 let plan = lazy
8406 .dump_plan(None, None, Some("<http://ex/p/07>"), None)
8407 .scan
8408 .expect("a known predicate plans a scan");
8409 assert!(
8410 reader.bytes_read() - before < 64 * 1024,
8411 "planning fetched {} B — it is supposed to read directories, not tiles",
8412 reader.bytes_read() - before
8413 );
8414 assert!(plan.tiles_admitted < plan.tiles_total);
8415 assert!(
8421 plan.tile_bytes * 5 < plan.section_bytes,
8422 "one predicate of twenty planned {} B of a {} B section",
8423 plan.tile_bytes,
8424 plan.section_bytes
8425 );
8426 }
8427
8428 #[test]
8429 fn query_quads_tags_every_graph() {
8430 let mut db = DictionaryBuilder::new();
8431 for (s, p, o) in [("Alice", "knows", "Bob"), ("Alice", "knows", "Dave")] {
8432 db.observe(s, p, o);
8433 }
8434 let dict = db.build();
8435 let mut def = GraphIndexBuilder::new();
8436 def.push(dict.encode("Alice", "knows", "Bob").unwrap());
8437 let mut g1 = GraphIndexBuilder::new();
8438 g1.push(dict.encode("Alice", "knows", "Dave").unwrap());
8439 let named = vec![("http://ex/g1".to_string(), g1.build())];
8440 let bytes = write_dataset(&dict, &def.build(), &named, true, &[], 0);
8441 let rete = Rete::open(&bytes).unwrap();
8442
8443 let quads = rete.query_quads(Some("Alice"), Some("knows"), None);
8445 assert_eq!(quads.len(), 2);
8446 assert_eq!(
8447 quads[0],
8448 (("Alice".into(), "knows".into(), "Bob".into()), None)
8449 );
8450 assert_eq!(
8451 quads[1],
8452 (
8453 ("Alice".into(), "knows".into(), "Dave".into()),
8454 Some("http://ex/g1".to_string())
8455 )
8456 );
8457
8458 assert!(rete.query_quads(Some("Nobody"), None, None).is_empty());
8460 }
8461
8462 #[test]
8463 fn pyramid_meta_round_trips_in_file() {
8464 let rete = Rete::open(&build_image()).unwrap();
8465 let pyr = rete.pyramid().expect("file has a pyramid");
8466 let total: u32 = pyr.summary.iter().map(|e| e.count).sum();
8468 assert_eq!(total, 3);
8469 assert!(!pyr.summary.is_empty());
8470 assert!(pyr.tiles.is_empty());
8471 }
8472
8473 #[test]
8474 fn schema_pyramid_round_trips_through_file_index_free() {
8475 use crate::reader::{CountingReader, SliceReader};
8476 let sub = "<http://www.w3.org/2000/01/rdf-schema#subClassOf>";
8477 let q = |s: &str, p: &str, o: &str| {
8478 (s.to_string(), p.to_string(), o.to_string(), None::<String>)
8479 };
8480 let quads = vec![
8482 q("<a>", RDF_TYPE, "<Astronomer>"),
8483 q("<b>", RDF_TYPE, "<Astronomer>"),
8484 q("<c>", RDF_TYPE, "<Person>"),
8485 q("<Astronomer>", sub, "<Scientist>"),
8486 q("<Scientist>", sub, "<Person>"),
8487 q("<Person>", sub, "<Agent>"),
8488 q("<a>", "<knows>", "<b>"),
8489 q("<b>", "<knows>", "<c>"),
8490 ];
8491 let (bytes, _) =
8492 crate::ingest::assemble_dataset_with_opts(quads, true, false, None, |_, _| Vec::new());
8493
8494 let rete = Rete::open(&bytes).unwrap();
8496 let pyr = rete.pyramid().expect("pyramid present");
8497 assert!(!pyr.level_rollups.is_empty(), "schema pyramid shipped");
8498 assert!(pyr
8499 .class_hierarchy
8500 .iter()
8501 .any(|n| n.class == "<Agent>" && n.depth == 0));
8502
8503 let r = CountingReader::new(SliceReader::new(&bytes));
8505 let view = SummaryView::open_ranged(&r).unwrap().unwrap();
8506 assert!(view.level_count() >= 2, "multi-level pyramid");
8507 let coarse = view.level_rollup(0).unwrap();
8508 assert!(
8509 coarse.classes.iter().any(|(c, _)| c == "<Agent>"),
8510 "coarsest level rolls up to the root Agent"
8511 );
8512 let h = Header::from_bytes(&bytes).unwrap();
8513 assert!(
8514 r.bytes_read() <= bytes.len() as u64 - h.root_dir_len,
8515 "summary read {} bytes; the {}-byte index section must be skipped",
8516 r.bytes_read(),
8517 h.root_dir_len
8518 );
8519 }
8520
8521 #[test]
8522 fn predicate_totals_from_summary_only() {
8523 use crate::reader::SliceReader;
8524 let bytes = build_image();
8526 let reader = SliceReader::new(&bytes);
8527 let view = SummaryView::open_ranged(&reader).unwrap().unwrap();
8528 assert_eq!(view.predicate_total("knows"), 2);
8529 assert_eq!(view.predicate_total("age"), 1);
8530 assert_eq!(view.predicate_total("missing"), 0);
8531 let totals = view.predicate_totals();
8532 assert_eq!(totals[0], ("knows".to_string(), 2)); }
8534
8535 #[test]
8536 fn query_patterns_resolve_to_terms() {
8537 let rete = Rete::open(&build_image()).unwrap();
8538
8539 assert_eq!(rete.query(None, None, None).len(), 3);
8541
8542 let mut alice = rete.query(Some("Alice"), None, None);
8544 alice.sort();
8545 assert_eq!(
8546 alice,
8547 vec![
8548 ("Alice".into(), "age".into(), "30".into()),
8549 ("Alice".into(), "knows".into(), "Bob".into()),
8550 ]
8551 );
8552
8553 assert_eq!(rete.query(None, Some("knows"), None).len(), 2);
8555
8556 assert_eq!(
8558 rete.query(Some("Bob"), Some("knows"), Some("Carol")),
8559 vec![("Bob".into(), "knows".into(), "Carol".into())]
8560 );
8561 assert!(rete.query(Some("Nobody"), None, None).is_empty());
8562 assert!(rete.query(None, Some("likes"), None).is_empty());
8563 }
8564
8565 #[test]
8566 fn query_provenance_reports_terms_ids_sections_and_index_choice() {
8567 let bytes = build_image();
8568 let rete = Rete::open(&bytes).unwrap();
8569
8570 let mut matches = rete.query_with_provenance(None, Some("knows"), None);
8571 matches.sort_by(|a, b| a.terms.cmp(&b.terms));
8572
8573 assert_eq!(matches.len(), 2);
8574 assert_eq!(
8575 matches[0].terms,
8576 ("Alice".into(), "knows".into(), "Bob".into())
8577 );
8578 assert_eq!(
8579 matches[0].ids,
8580 rete.dictionary().encode("Alice", "knows", "Bob").unwrap()
8581 );
8582 assert_eq!(matches[0].graph.as_deref(), None);
8583 assert_eq!(
8584 matches[0].matched_pattern,
8585 (None, Some(matches[0].ids.1), None)
8586 );
8587 assert_eq!(
8588 matches[0].index_permutation,
8589 crate::index::IndexPermutation::Pos
8590 );
8591
8592 let h = rete.header();
8593 assert_eq!(matches[0].dictionary_range.offset, h.dictionary_offset);
8594 assert_eq!(matches[0].dictionary_range.len, h.dictionary_len);
8595 assert_eq!(matches[0].index_range.offset, h.root_dir_offset);
8596 assert_eq!(matches[0].index_range.len, h.root_dir_len);
8597 assert!(
8598 matches[0].index_section_range.offset > h.root_dir_offset,
8599 "POS is section 1, so its payload starts after the container header and SPO payload"
8600 );
8601 assert!(matches[0].index_section_range.len > 0);
8602 assert!(matches[0].index_section_range.end() <= matches[0].index_range.end());
8603 assert!(matches[0].index_section_range.len < matches[0].index_range.len);
8604 assert_eq!(
8605 matches[0].pyramid_range.as_ref().map(|r| (r.offset, r.len)),
8606 Some((h.pyramid_meta_offset, h.pyramid_meta_len))
8607 );
8608 let tile_range = matches[0].tile_range.expect("tiled file reports a tile");
8611 assert!(matches[0]
8612 .tile
8613 .as_deref()
8614 .unwrap()
8615 .starts_with(matches[0].index_permutation.name()));
8616 assert!(matches[0].index_section_range.offset <= tile_range.offset);
8617 assert!(tile_range.end() <= matches[0].index_section_range.end());
8618 }
8619
8620 fn build_labeled(n: usize) -> Vec<u8> {
8624 const LABEL: &str = "<http://www.w3.org/2000/01/rdf-schema#label>";
8625 const WORDS: &[&str] = &[
8626 "alanine",
8627 "benzene",
8628 "glucose",
8629 "dextrose",
8630 "ethanol",
8631 "formate",
8632 "heptane",
8633 "isoleucine",
8634 ];
8635 let triples: Vec<(String, String, String)> = (0..n)
8636 .flat_map(|i| {
8637 let s = format!("<http://ex/e{i}>");
8638 let w = WORDS[i % WORDS.len()];
8639 [
8640 (s.clone(), LABEL.to_string(), format!("\"{w}-{i:06}\"")),
8641 (
8642 s,
8643 "<http://ex/p>".to_string(),
8644 format!("<http://ex/c{}>", i % 64),
8645 ),
8646 ]
8647 })
8648 .collect();
8649 let mut db = DictionaryBuilder::new();
8650 for (s, p, o) in &triples {
8651 db.observe(s, p, o);
8652 }
8653 let dict = db.build();
8654 let ids: Vec<(u32, u32, u32)> = triples
8655 .iter()
8656 .map(|(s, p, o)| dict.encode(s, p, o).unwrap())
8657 .collect();
8658 let mut ib = GraphIndexBuilder::new();
8659 for &t in &ids {
8660 ib.push(t);
8661 }
8662 let (meta, levels) = build_pyramid_meta(&dict, &ids, DEFAULT_TILE_BUDGET);
8663 write_file(&dict, &ib.build(), false, &meta, levels)
8664 }
8665
8666 #[test]
8667 fn prefix_search_matches_a_filter_scan() {
8668 let bytes = build_labeled(800);
8671 let rete = Rete::open(&bytes).unwrap();
8672 let idx_subjects: std::collections::BTreeSet<String> = rete
8673 .prefix_search("glucose", 10_000)
8674 .into_iter()
8675 .map(|(_label, subject)| subject)
8676 .collect();
8677 let q = "SELECT ?s WHERE { ?s <http://www.w3.org/2000/01/rdf-schema#label> ?l \
8679 FILTER(STRSTARTS(LCASE(?l), \"glucose\")) }";
8680 let crate::QueryOutput::Select(_, rows) = crate::eval_query(&rete, q).unwrap() else {
8681 panic!("expected SELECT");
8682 };
8683 let scan_subjects: std::collections::BTreeSet<String> =
8684 rows.iter().map(|r| r.get("s").cloned().unwrap()).collect();
8685 assert_eq!(idx_subjects, scan_subjects, "index agrees with the scan");
8686 assert_eq!(
8687 idx_subjects.len(),
8688 100,
8689 "800/8 words = 100 glucose-* labels"
8690 );
8691 }
8692
8693 #[test]
8697 #[ignore]
8698 fn bench_prefix_search_vs_filter_scan() {
8699 use std::time::Instant;
8700 let n = 6000; let bytes = build_labeled(n);
8702 let rete = Rete::open(&bytes).unwrap();
8703 let q = "SELECT ?s WHERE { ?s <http://www.w3.org/2000/01/rdf-schema#label> ?l \
8704 FILTER(STRSTARTS(LCASE(?l), \"glucose\")) }";
8705 let reps = 200;
8706 let idx_n = rete.prefix_search("glucose", 100_000).len();
8707 let t = Instant::now();
8708 for _ in 0..reps {
8709 std::hint::black_box(rete.prefix_search("glucose", 100_000));
8710 }
8711 let idx_ms = t.elapsed().as_secs_f64() * 1000.0 / reps as f64;
8712 let t = Instant::now();
8713 for _ in 0..reps {
8714 let _ = std::hint::black_box(crate::eval_query(&rete, q).unwrap());
8715 }
8716 let scan_ms = t.elapsed().as_secs_f64() * 1000.0 / reps as f64;
8717 println!(
8718 "label prefix search over {n} labeled subjects ({idx_n} matches): \
8719 index {idx_ms:.4} ms vs FILTER scan {scan_ms:.3} ms ({:.0}× faster)",
8720 scan_ms / idx_ms
8721 );
8722 }
8723
8724 #[test]
8728 #[ignore]
8729 fn bench_text_search_vs_contains_scan() {
8730 use std::time::Instant;
8731 const LABEL: &str = "<http://www.w3.org/2000/01/rdf-schema#label>";
8732 const WORDS: &[&str] = &[
8733 "alanine",
8734 "benzene",
8735 "glucose",
8736 "dextrose",
8737 "ethanol",
8738 "formate",
8739 "heptane",
8740 "isoleucine",
8741 ];
8742 let n = 6000;
8743 let triples: Vec<(String, String, String)> = (0..n)
8744 .map(|i| {
8745 (
8746 format!("<http://ex/e{i}>"),
8747 LABEL.to_string(),
8748 format!("\"{} sample number {i:06}\"", WORDS[i % WORDS.len()]),
8749 )
8750 })
8751 .collect();
8752 let mut db = DictionaryBuilder::new();
8753 for (s, p, o) in &triples {
8754 db.observe(s, p, o);
8755 }
8756 let dict = db.build();
8757 let ids: Vec<(u32, u32, u32)> = triples
8758 .iter()
8759 .map(|(s, p, o)| dict.encode(s, p, o).unwrap())
8760 .collect();
8761 let mut ib = GraphIndexBuilder::new();
8762 for &t in &ids {
8763 ib.push(t);
8764 }
8765 let ti = compute_text_index(&dict, &ids);
8766 let bytes = write_dataset_with_metadata(&dict, &ib.build(), &[], false, &[], 0, &[], &ti);
8767 let rete = Rete::open(&bytes).unwrap();
8768
8769 let q = "SELECT ?s WHERE { ?s <http://www.w3.org/2000/01/rdf-schema#label> ?l \
8770 FILTER(CONTAINS(LCASE(?l), \"glucose\")) }";
8771 let reps = 200;
8772 let idx_n = rete.text_search(&["glucose"], None, 100_000).len();
8773 let t = Instant::now();
8774 for _ in 0..reps {
8775 std::hint::black_box(rete.text_search(&["glucose"], None, 100_000));
8776 }
8777 let idx_ms = t.elapsed().as_secs_f64() * 1000.0 / reps as f64;
8778 let t = Instant::now();
8779 for _ in 0..reps {
8780 let _ = std::hint::black_box(crate::eval_query(&rete, q).unwrap());
8781 }
8782 let scan_ms = t.elapsed().as_secs_f64() * 1000.0 / reps as f64;
8783 println!(
8784 "text search over {n} literals ({idx_n} matches): \
8785 index {idx_ms:.4} ms vs FILTER(CONTAINS) scan {scan_ms:.3} ms ({:.0}× faster)",
8786 scan_ms / idx_ms
8787 );
8788 }
8789
8790 #[test]
8794 #[ignore = "operational tool, driven by RETE_DEBUG_* env vars"]
8795 fn debug_bound_po_routing() {
8796 struct FR(std::fs::File);
8797 impl crate::RangeReader for FR {
8798 fn len(&self) -> u64 {
8799 self.0.metadata().map(|m| m.len()).unwrap_or(0)
8800 }
8801 fn read_at(&self, offset: u64, len: u64) -> std::io::Result<Vec<u8>> {
8802 use std::os::unix::fs::FileExt;
8803 let mut buf = vec![0u8; len as usize];
8804 self.0.read_exact_at(&mut buf, offset)?;
8805 Ok(buf)
8806 }
8807 }
8808 let path = std::env::var("RETE_DEBUG_FILE").expect("RETE_DEBUG_FILE");
8809 let p_iri = std::env::var("RETE_DEBUG_P").expect("RETE_DEBUG_P");
8810 let o_iri = std::env::var("RETE_DEBUG_O").expect("RETE_DEBUG_O");
8811 let rete =
8812 Rete::open_ranged_lazy(std::sync::Arc::new(FR(std::fs::File::open(&path).unwrap())))
8813 .unwrap();
8814 let pid = rete.dict.predicate_id(&p_iri).expect("p resolves");
8815 let oid = rete.dict.object_id(&o_iri).expect("o resolves");
8816 eprintln!("pid={pid} oid={oid}");
8817 let pattern = (None, Some(pid), Some(oid));
8818 let perm = GraphIndex::best_permutation(pattern);
8819 eprintln!("best_permutation = {}", perm.name());
8820 let si = perm.section_index();
8821 let tiles = &rete.index.sections[si];
8822 eprintln!("section {} tiles = {}", perm.name(), tiles.len());
8823 let [pa, pb, pc] = perm.order_pattern(pattern);
8824 eprintln!("permuted pattern pa={pa:?} pb={pb:?} pc={pc:?}");
8825 let (start, end) = rete.index.tile_span(si, pa);
8826 eprintln!("tile_span = [{start}, {end}) -> {} tiles", end - start);
8827 let mut admitted = 0usize;
8828 for (ti, t) in tiles.iter().enumerate().take(end).skip(start) {
8829 if t.syn_admits(pb, pc) {
8830 admitted += 1;
8831 if admitted <= 10 {
8832 let (lo, hi) = t.leading_range();
8833 eprintln!(" admit tile {ti}: a=[{lo},{hi}] syn={:?}", t.syn);
8834 }
8835 }
8836 }
8837 eprintln!("admitted {admitted} tile(s) by synopsis");
8838 let n = rete.index.scan_iter(pattern).count();
8839 eprintln!("scan_iter matches = {n}");
8840 let hi_res = rete.query(None, Some(&p_iri), Some(&o_iri));
8841 eprintln!("high-level query matches = {}", hi_res.len());
8842 }
8843
8844 #[test]
8852 fn layout_header_label_matches_its_span() {
8853 let mut db = DictionaryBuilder::new();
8854 let mut triples = Vec::new();
8855 for i in 0..24u32 {
8856 let (s, p, o) = (
8857 format!("<http://ex/s{}>", i % 5),
8858 format!("<http://ex/p{}>", i % 3),
8859 format!("<http://ex/o{i}>"),
8860 );
8861 db.observe(&s, &p, &o);
8862 triples.push((s, p, o));
8863 }
8864 let dict = db.build();
8865 let ids: Vec<(u32, u32, u32)> = triples
8866 .iter()
8867 .map(|(s, p, o)| dict.encode(s, p, o).unwrap())
8868 .collect();
8869 let index = GraphIndexBuilder::from_triples(ids).build();
8870 let image = write_dataset_with_metadata(
8871 &dict,
8872 &index,
8873 &[],
8874 false,
8875 &[],
8876 0,
8877 br#"{"name":"layout-label-fixture"}"#,
8878 &[],
8879 );
8880 let rete = Rete::open(&image).unwrap();
8881
8882 let layout = rete.file_layout();
8883 let header = layout
8884 .iter()
8885 .find(|s| s.kind == "header")
8886 .expect("every layout starts with a header segment");
8887
8888 assert_eq!(
8889 header.offset, 0,
8890 "the header is the first thing in the file"
8891 );
8892 assert_eq!(
8893 header.len, HEADER_LEN as u64,
8894 "the header segment must span exactly HEADER_LEN"
8895 );
8896
8897 let digits: String = header
8900 .label
8901 .chars()
8902 .skip_while(|c| !c.is_ascii_digit())
8903 .take_while(|c| c.is_ascii_digit())
8904 .collect();
8905 assert!(
8906 !digits.is_empty(),
8907 "header label {:?} states no size at all",
8908 header.label
8909 );
8910 assert_eq!(
8911 digits.parse::<u64>().unwrap(),
8912 header.len,
8913 "header label {:?} disagrees with the {}-byte span it describes",
8914 header.label,
8915 header.len
8916 );
8917
8918 assert!(
8921 layout.windows(2).all(|w| w[0].offset <= w[1].offset),
8922 "file_layout must return segments sorted by offset"
8923 );
8924 }
8925}