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, 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 CODEC_ZSTD => {
397 use std::io::Read;
398 let mut dec = ruzstd::StreamingDecoder::new(bytes)
399 .map_err(|e| FileError::Decompress(std::io::Error::other(e.to_string())))?;
400 let mut out = Vec::new();
401 dec.read_to_end(&mut out).map_err(FileError::Decompress)?;
402 Ok(out)
403 }
404 other => Err(FileError::UnknownCodec(other)),
405 }
406}
407
408const TILE_COALESCE_GAP: u64 = 4096;
412
413const DICT_COALESCE_GAP: u64 = 64 * 1024;
419
420fn read_coalesced<R: RangeReader + ?Sized>(
426 reader: &R,
427 ranges: &[ByteRange],
428 gap: u64,
429) -> Option<Vec<Vec<u8>>> {
430 let mut spans: Vec<(u64, u64)> = Vec::new();
433 let mut span_of: Vec<usize> = Vec::with_capacity(ranges.len());
434 let mut i = 0;
435 while i < ranges.len() {
436 let start = ranges[i].offset;
437 let mut end = ranges[i].offset.checked_add(ranges[i].len)?;
438 let mut j = i + 1;
439 while j < ranges.len() {
440 let r = &ranges[j];
441 if r.offset < end || r.offset - end > gap {
442 break;
443 }
444 end = r.offset.checked_add(r.len)?;
445 j += 1;
446 }
447 let si = spans.len();
448 spans.push((start, end - start));
449 for _ in i..j {
450 span_of.push(si);
451 }
452 i = j;
453 }
454 let blobs = reader.read_many(&spans).ok()?;
455 if blobs.len() != spans.len() {
456 return None;
457 }
458 let mut out = Vec::with_capacity(ranges.len());
459 for (k, r) in ranges.iter().enumerate() {
460 let (span_start, _) = spans[span_of[k]];
461 let blob = &blobs[span_of[k]];
462 let lo = (r.offset - span_start) as usize;
463 let hi = lo.checked_add(r.len as usize)?;
464 out.push(blob.get(lo..hi)?.to_vec());
465 }
466 Some(out)
467}
468
469fn content_hash(parts: &[&[u8]]) -> [u8; 16] {
472 let mut h = blake3::Hasher::new();
473 for p in parts {
474 h.update(p);
475 }
476 let mut out = [0u8; 16];
477 out.copy_from_slice(&h.finalize().as_bytes()[..16]);
478 out
479}
480
481pub type TermTriple = (String, String, String);
483
484#[derive(Debug, Clone)]
488pub struct LayoutSegment {
489 pub kind: &'static str,
490 pub label: String,
491 pub offset: u64,
492 pub len: u64,
493}
494
495#[derive(Debug, Clone, Copy, PartialEq, Eq)]
497pub struct ByteRange {
498 pub offset: u64,
499 pub len: u64,
500}
501
502impl ByteRange {
503 pub fn end(self) -> u64 {
504 self.offset + self.len
505 }
506}
507
508#[derive(Debug, Clone, PartialEq, Eq)]
510pub struct TripleProvenance {
511 pub terms: TermTriple,
513 pub ids: Triple,
515 pub graph: Option<String>,
517 pub matched_pattern: Pattern,
519 pub index_permutation: IndexPermutation,
521 pub dictionary_range: ByteRange,
523 pub index_range: ByteRange,
525 pub index_section_range: ByteRange,
528 pub pyramid_range: Option<ByteRange>,
530 pub tile: Option<String>,
534 pub tile_range: Option<ByteRange>,
537}
538
539fn encode_container(sections: &[&[u8]], codec: u8) -> Vec<u8> {
543 let mut out = Vec::new();
544 write_uvarint(&mut out, sections.len() as u64);
545 for s in sections {
546 let payload = compress(codec, s);
547 write_uvarint(&mut out, payload.len() as u64);
548 out.extend_from_slice(&payload);
549 }
550 out
551}
552
553fn decode_container(bytes: &[u8], codec: u8) -> Result<Vec<Vec<u8>>, FileError> {
555 let (n, mut pos) = read_uvarint(bytes).ok_or(FileError::Container("truncated count"))?;
556 let mut out = Vec::with_capacity((n as usize).min(bytes.len()));
560 for _ in 0..n {
561 let (len, used) =
562 read_uvarint(&bytes[pos..]).ok_or(FileError::Container("truncated length"))?;
563 pos += used;
564 let end = pos + len as usize;
565 if end > bytes.len() {
566 return Err(FileError::Container("section overruns buffer"));
567 }
568 out.push(decompress(codec, &bytes[pos..end])?);
569 pos = end;
570 }
571 Ok(out)
572}
573
574fn checked_end(off: u64, len: u64) -> Result<u64, FileError> {
575 off.checked_add(len)
576 .ok_or(FileError::Container("section range overflows"))
577}
578
579const DICT_CHUNK_BUDGET: usize = 64 * 1024;
582
583fn encode_chunked_dict_section(raw: &[u8], codec: u8) -> Vec<u8> {
590 let meta = crate::dict::parse_meta(raw).unwrap_or(crate::dict::SectionMeta {
591 term_count: 0,
592 restart_interval: 1,
593 restart_offsets: Vec::new(),
594 });
595 let body_start = meta
596 .restart_offsets
597 .first()
598 .copied()
599 .unwrap_or(raw.len() as u64);
600 let header = &raw[..(body_start.min(raw.len() as u64)) as usize];
601
602 let n_runs = meta.restart_offsets.len();
604 let mut bounds: Vec<(usize, u64, u64)> = Vec::new(); let mut r = 0;
606 while r < n_runs {
607 let start = meta.restart_offsets[r];
608 let mut r2 = r + 1;
609 while r2 < n_runs && meta.restart_offsets[r2] - start < DICT_CHUNK_BUDGET as u64 {
610 r2 += 1;
611 }
612 let end = if r2 < n_runs {
613 meta.restart_offsets[r2]
614 } else {
615 raw.len() as u64
616 };
617 bounds.push((r, start, end));
618 r = r2;
619 }
620
621 let compressed: Vec<Vec<u8>> = bounds
622 .iter()
623 .map(|&(_, s, e)| compress(codec, &raw[s as usize..e as usize]))
624 .collect();
625 let mut out = Vec::new();
626 write_uvarint(&mut out, header.len() as u64);
627 out.extend_from_slice(header);
628 write_uvarint(&mut out, bounds.len() as u64);
629 let mut prev_run = 0usize;
630 for (&(first_run, start, _), comp) in bounds.iter().zip(&compressed) {
631 let first_term = crate::dict::run_first_term(raw, start as usize).unwrap_or_default();
632 write_uvarint(&mut out, (first_run - prev_run) as u64);
633 write_uvarint(&mut out, first_term.len() as u64);
634 out.extend_from_slice(&first_term);
635 write_uvarint(&mut out, comp.len() as u64);
636 prev_run = first_run;
637 }
638 for comp in &compressed {
639 out.extend_from_slice(comp);
640 }
641 out
642}
643
644struct DictChunkEntry {
647 first_run: usize,
648 first_term: Vec<u8>,
649 body_start: u64,
650 start: u64,
651 end: u64,
652}
653
654fn parse_chunked_dict_dir(
658 bytes: &[u8],
659 total_len: u64,
660) -> Result<(crate::dict::SectionMeta, Vec<DictChunkEntry>), FileError> {
661 let mut pos = 0usize;
662 let take = |pos: &mut usize| -> Result<u64, FileError> {
663 let (v, n) = read_uvarint(bytes.get(*pos..).unwrap_or(&[]))
664 .ok_or(FileError::Container("truncated dict chunk directory"))?;
665 *pos += n;
666 Ok(v)
667 };
668 let header_len = take(&mut pos)? as usize;
669 let header = bytes
670 .get(pos..pos.saturating_add(header_len))
671 .ok_or(FileError::Container("truncated dict header"))?;
672 let meta = crate::dict::parse_meta(header)
673 .map_err(|_| FileError::Container("malformed dict header"))?;
674 pos += header_len;
675
676 let num_chunks = take(&mut pos)? as usize;
677 let mut entries = Vec::with_capacity(num_chunks.min(bytes.len()));
678 let mut lens = Vec::with_capacity(num_chunks.min(bytes.len()));
679 let mut prev_run = 0usize;
680 for _ in 0..num_chunks {
681 let drun = take(&mut pos)? as usize;
682 let tlen = take(&mut pos)? as usize;
683 let term = bytes
684 .get(pos..pos.saturating_add(tlen))
685 .ok_or(FileError::Container("truncated dict chunk first term"))?
686 .to_vec();
687 pos += tlen;
688 let clen = take(&mut pos)?;
689 let first_run = prev_run + drun;
690 let body_start = meta
691 .restart_offsets
692 .get(first_run)
693 .copied()
694 .ok_or(FileError::Container("dict chunk run out of range"))?;
695 entries.push(DictChunkEntry {
696 first_run,
697 first_term: term,
698 body_start,
699 start: 0,
700 end: 0,
701 });
702 lens.push(clen);
703 prev_run = first_run;
704 }
705 let mut start = pos as u64;
706 for (e, len) in entries.iter_mut().zip(lens) {
707 let end = start
708 .checked_add(len)
709 .filter(|&e| e <= total_len)
710 .ok_or(FileError::Container("dict chunk overruns section"))?;
711 e.start = start;
712 e.end = end;
713 start = end;
714 }
715 Ok((meta, entries))
716}
717
718fn read_dict_dir_ranged<R: RangeReader>(
730 reader: &R,
731 section: ByteRange,
732) -> Result<(crate::dict::SectionMeta, Vec<DictChunkEntry>), FileError> {
733 let total = section.len;
734 let init = 8192.min(total); let head = reader.read_at(section.offset, init)?;
742 let (header_len, n0) =
743 read_uvarint(&head).ok_or(FileError::Container("truncated dict header len"))?;
744 let hbase = n0; let (term_count, n1) = read_uvarint(head.get(hbase..).unwrap_or(&[]))
746 .ok_or(FileError::Container("truncated dict term_count"))?;
747 let (restart_interval, _n2) = read_uvarint(head.get(hbase + n1..).unwrap_or(&[]))
748 .ok_or(FileError::Container("truncated dict interval"))?;
749 if restart_interval == 0 {
750 return Err(FileError::Container("zero restart interval"));
751 }
752 let dir_start = (hbase as u64)
755 .checked_add(header_len)
756 .filter(|&d| d <= total)
757 .ok_or(FileError::Container("dict header overruns section"))?;
758 let dir_total = total - dir_start;
759 let meta = crate::dict::SectionMeta {
760 term_count: term_count as u32,
761 restart_interval: restart_interval as u32,
762 restart_offsets: Vec::new(),
763 };
764 let finish = |mut entries: Vec<DictChunkEntry>| {
765 for e in &mut entries {
766 e.start += dir_start; e.end += dir_start;
768 }
769 (meta.clone(), entries)
770 };
771 if dir_start < head.len() as u64 {
774 if let Ok(entries) = parse_chunk_dir_only(&head[dir_start as usize..], dir_total) {
775 return Ok(finish(entries));
776 }
777 }
778 let mut prefetch = 4096u64.min(dir_total).max(1);
780 loop {
781 let dir = reader.read_at(section.offset + dir_start, prefetch)?;
782 match parse_chunk_dir_only(&dir, dir_total) {
783 Ok(entries) => return Ok(finish(entries)),
784 Err(_) if prefetch < dir_total => prefetch = prefetch.saturating_mul(2).min(dir_total),
785 Err(e) => return Err(e),
786 }
787 }
788}
789
790fn parse_chunk_dir_only(dir: &[u8], dir_total: u64) -> Result<Vec<DictChunkEntry>, FileError> {
797 let mut pos = 0usize;
798 let take = |pos: &mut usize| -> Result<u64, FileError> {
799 let (v, n) = read_uvarint(dir.get(*pos..).unwrap_or(&[]))
800 .ok_or(FileError::Container("truncated dict chunk directory"))?;
801 *pos += n;
802 Ok(v)
803 };
804 let num_chunks = take(&mut pos)? as usize;
805 let mut entries = Vec::with_capacity(num_chunks.min(dir.len()));
806 let mut lens = Vec::with_capacity(num_chunks.min(dir.len()));
807 let mut prev_run = 0usize;
808 for _ in 0..num_chunks {
809 let drun = take(&mut pos)? as usize;
810 let tlen = take(&mut pos)? as usize;
811 let term = dir
812 .get(pos..pos.saturating_add(tlen))
813 .ok_or(FileError::Container("truncated dict chunk first term"))?
814 .to_vec();
815 pos += tlen;
816 let clen = take(&mut pos)?;
817 let first_run = prev_run + drun;
818 entries.push(DictChunkEntry {
819 first_run,
820 first_term: term,
821 body_start: 0,
822 start: 0,
823 end: 0,
824 });
825 lens.push(clen);
826 prev_run = first_run;
827 }
828 let mut start = pos as u64;
829 for (e, len) in entries.iter_mut().zip(lens) {
830 let end = start
831 .checked_add(len)
832 .filter(|&e| e <= dir_total)
833 .ok_or(FileError::Container("dict chunk overruns section"))?;
834 e.start = start;
835 e.end = end;
836 start = end;
837 }
838 Ok(entries)
839}
840
841fn decode_chunked_dict_section(
845 payload: &[u8],
846 codec: u8,
847) -> Result<crate::dict::ChunkedSection, FileError> {
848 let (meta, entries) = parse_chunked_dict_dir(payload, payload.len() as u64)?;
849 let chunks = entries
850 .into_iter()
851 .map(|e| {
852 Ok(crate::dict::SectionChunk::resident(
853 e.first_run,
854 e.first_term,
855 e.body_start,
856 decompress(codec, &payload[e.start as usize..e.end as usize])?,
857 ))
858 })
859 .collect::<Result<Vec<_>, FileError>>()?;
860 Ok(crate::dict::ChunkedSection::from_parts(meta, chunks, None))
861}
862
863fn decode_dictionary_container(bytes: &[u8], codec: u8) -> Result<Dictionary, FileError> {
864 let dsecs = decode_container(bytes, CODEC_NONE)?;
865 if dsecs.len() != 4 {
866 return Err(FileError::Container("expected 4 dictionary sections"));
867 }
868 let mut sections = Vec::with_capacity(4);
869 for sec in &dsecs {
870 sections.push(decode_chunked_dict_section(sec, codec)?);
871 }
872 let arr: [crate::dict::ChunkedSection; 4] = sections
873 .try_into()
874 .map_err(|_| FileError::Container("expected 4 dictionary sections"))?;
875 Ok(Dictionary::from_chunked_sections(arr))
876}
877
878fn encode_tiled_section(tiles: &[crate::index::Tile], codec: u8) -> Vec<u8> {
884 #[cfg(feature = "parallel")]
888 let compressed: Vec<Vec<u8>> = {
889 use rayon::prelude::*;
890 tiles
891 .par_iter()
892 .map(|t| compress(codec, t.bytes()))
893 .collect()
894 };
895 #[cfg(not(feature = "parallel"))]
896 let compressed: Vec<Vec<u8>> = tiles.iter().map(|t| compress(codec, t.bytes())).collect();
897 let mut out = Vec::new();
898 write_uvarint(&mut out, tiles.len() as u64);
899 let mut prev_min = 0u32;
900 for (tile, comp) in tiles.iter().zip(&compressed) {
901 let (min_a, max_a) = tile.leading_range();
902 write_uvarint(&mut out, (min_a - prev_min) as u64);
903 write_uvarint(&mut out, (max_a - min_a) as u64);
904 write_uvarint(&mut out, comp.len() as u64);
905 prev_min = min_a;
906 }
907 for comp in &compressed {
908 out.extend_from_slice(comp);
909 }
910 for tile in tiles {
918 let (min_b, max_b, min_c, max_c) = match crate::triples::TripleBlock::parse(tile.bytes()) {
919 Ok(b) => {
920 let z = b.zone();
921 (z.min_b, z.max_b, z.min_c, z.max_c)
922 }
923 Err(_) => (0, u32::MAX, 0, u32::MAX),
924 };
925 write_uvarint(&mut out, min_b as u64);
926 write_uvarint(&mut out, (max_b - min_b) as u64);
927 write_uvarint(&mut out, min_c as u64);
928 write_uvarint(&mut out, (max_c - min_c) as u64);
929 }
930 out
931}
932
933struct TileDirEntry {
936 min_a: u32,
937 max_a: u32,
938 start: u64,
939 end: u64,
940}
941
942type TileSynopsis = (u32, u32, u32, u32);
944
945fn parse_tile_synopsis(
953 payload: &[u8],
954 trailer_start: usize,
955 num_tiles: usize,
956) -> Option<Vec<TileSynopsis>> {
957 let mut pos = trailer_start;
958 let take = |pos: &mut usize| -> Option<u32> {
959 let (v, n) = read_uvarint(payload.get(*pos..)?)?;
960 *pos += n;
961 u32::try_from(v).ok()
962 };
963 let mut out = Vec::with_capacity(num_tiles.min(payload.len()));
964 for _ in 0..num_tiles {
965 let min_b = take(&mut pos)?;
966 let max_b = min_b.checked_add(take(&mut pos)?)?;
967 let min_c = take(&mut pos)?;
968 let max_c = min_c.checked_add(take(&mut pos)?)?;
969 out.push((min_b, max_b, min_c, max_c));
970 }
971 Some(out)
972}
973
974fn parse_tile_directory(bytes: &[u8], total_len: u64) -> Result<Vec<TileDirEntry>, FileError> {
979 let mut pos = 0usize;
980 let take = |pos: &mut usize| -> Result<u64, FileError> {
981 let (v, n) = read_uvarint(bytes.get(*pos..).unwrap_or(&[]))
982 .ok_or(FileError::Container("truncated tile directory"))?;
983 *pos += n;
984 Ok(v)
985 };
986 let num_tiles = take(&mut pos)? as usize;
987 let mut entries = Vec::with_capacity(num_tiles.min(bytes.len()));
988 let mut prev_min = 0u32;
989 let mut lens = Vec::with_capacity(num_tiles.min(bytes.len()));
990 for _ in 0..num_tiles {
991 let dmin = take(&mut pos)? as u32;
992 let span = take(&mut pos)? as u32;
993 let len = take(&mut pos)?;
994 let min_a = prev_min.wrapping_add(dmin);
995 entries.push(TileDirEntry {
996 min_a,
997 max_a: min_a.wrapping_add(span),
998 start: 0,
999 end: 0,
1000 });
1001 lens.push(len);
1002 prev_min = min_a;
1003 }
1004 let mut start = pos as u64;
1005 for (e, len) in entries.iter_mut().zip(lens) {
1006 let end = start
1007 .checked_add(len)
1008 .filter(|&e| e <= total_len)
1009 .ok_or(FileError::Container("tile overruns section"))?;
1010 e.start = start;
1011 e.end = end;
1012 start = end;
1013 }
1014 Ok(entries)
1015}
1016
1017fn read_tile_directory_ranged<R: RangeReader>(
1021 reader: &R,
1022 section: ByteRange,
1023) -> Result<Vec<TileDirEntry>, FileError> {
1024 let total = section.len;
1025 let mut prefetch = 4096u64.min(total);
1026 loop {
1027 let prefix = reader.read_at(section.offset, prefetch)?;
1028 match parse_tile_directory(&prefix, total) {
1029 Ok(dir) => return Ok(dir),
1030 Err(_) if prefetch < total => prefetch = prefetch.saturating_mul(2).min(total),
1031 Err(e) => return Err(e),
1032 }
1033 }
1034}
1035
1036fn read_tile_synopsis_ranged<R: RangeReader>(
1043 reader: &R,
1044 section: ByteRange,
1045 dir: &[TileDirEntry],
1046) -> Vec<Option<TileSynopsis>> {
1047 let n = dir.len();
1048 let none = vec![None; n];
1049 let trailer_start = dir.iter().map(|e| e.end).max().unwrap_or(0);
1050 let total = section.len;
1051 if n == 0 || trailer_start >= total {
1052 return none; }
1054 let trailer_len = total - trailer_start;
1055 let Ok(bytes) = reader.read_at(section.offset + trailer_start, trailer_len) else {
1056 return none;
1057 };
1058 match parse_tile_synopsis(&bytes, 0, n) {
1059 Some(v) => v.into_iter().map(Some).collect(),
1060 None => none,
1061 }
1062}
1063
1064fn tile_file_ranges(
1068 index_bytes: &[u8],
1069 container_offset: u64,
1070 section_ranges: &[ByteRange; NUM_PERMS],
1071) -> [Vec<(u32, u32, ByteRange)>; NUM_PERMS] {
1072 let mut out: [Vec<(u32, u32, ByteRange)>; NUM_PERMS] = Default::default();
1073 for (section, range) in out.iter_mut().zip(section_ranges) {
1074 let start = (range.offset - container_offset) as usize;
1075 let Some(payload) = index_bytes.get(start..start + range.len as usize) else {
1076 continue;
1077 };
1078 if let Ok(dir) = parse_tile_directory(payload, payload.len() as u64) {
1079 *section = dir
1080 .into_iter()
1081 .map(|e| {
1082 (
1083 e.min_a,
1084 e.max_a,
1085 ByteRange {
1086 offset: range.offset + e.start,
1087 len: (e.end - e.start),
1088 },
1089 )
1090 })
1091 .collect();
1092 }
1093 }
1094 out
1095}
1096
1097fn decode_tiled_section(payload: &[u8], codec: u8) -> Result<Vec<(u32, u32, Vec<u8>)>, FileError> {
1100 parse_tile_directory(payload, payload.len() as u64)?
1101 .into_iter()
1102 .map(|e| {
1103 Ok((
1104 e.min_a,
1105 e.max_a,
1106 decompress(codec, &payload[e.start as usize..e.end as usize])?,
1107 ))
1108 })
1109 .collect()
1110}
1111
1112fn decode_index_container(bytes: &[u8], codec: u8) -> Result<GraphIndex, FileError> {
1115 let mut isecs = decode_container(bytes, CODEC_NONE)?;
1116 if isecs.len() != NUM_PERMS {
1117 return Err(FileError::Container("expected 6 permutation sections"));
1118 }
1119 let mut sections: [Vec<(u32, u32, Vec<u8>)>; NUM_PERMS] = Default::default();
1120 for (i, sec) in isecs.iter_mut().enumerate() {
1121 sections[i] = decode_tiled_section(sec, codec)?;
1122 }
1123 Ok(GraphIndex::from_tiles(sections))
1124}
1125
1126fn container_section_payload_ranges(
1127 bytes: &[u8],
1128 container_offset: u64,
1129 expected_sections: usize,
1130) -> Result<Vec<ByteRange>, FileError> {
1131 let (section_count, mut pos) =
1132 read_uvarint(bytes).ok_or(FileError::Container("truncated count"))?;
1133 let section_count = usize::try_from(section_count)
1134 .map_err(|_| FileError::Container("section count too large"))?;
1135 if section_count != expected_sections {
1136 return Err(FileError::Container("unexpected section count"));
1137 }
1138
1139 let mut ranges = Vec::with_capacity(section_count);
1140 for _ in 0..section_count {
1141 let remaining = bytes
1142 .get(pos..)
1143 .ok_or(FileError::Container("truncated length"))?;
1144 let (payload_len, used) =
1145 read_uvarint(remaining).ok_or(FileError::Container("truncated length"))?;
1146 pos = pos
1147 .checked_add(used)
1148 .ok_or(FileError::Container("section range overflows"))?;
1149 let payload_len_usize = usize::try_from(payload_len)
1150 .map_err(|_| FileError::Container("section length too large"))?;
1151 let payload_end = pos
1152 .checked_add(payload_len_usize)
1153 .ok_or(FileError::Container("section range overflows"))?;
1154 if payload_end > bytes.len() {
1155 return Err(FileError::Container("section overruns buffer"));
1156 }
1157 ranges.push(ByteRange {
1158 offset: checked_end(container_offset, pos as u64)?,
1159 len: payload_len,
1160 });
1161 pos = payload_end;
1162 }
1163
1164 Ok(ranges)
1165}
1166
1167fn decode_index_section_ranges(
1168 bytes: &[u8],
1169 container_offset: u64,
1170) -> Result<[ByteRange; NUM_PERMS], FileError> {
1171 let ranges = container_section_payload_ranges(bytes, container_offset, NUM_PERMS)?;
1172 ranges
1173 .try_into()
1174 .map_err(|_| FileError::Container("expected 6 permutation blocks"))
1175}
1176
1177fn read_uvarint_at<R: RangeReader>(
1178 reader: &R,
1179 absolute_offset: u64,
1180 container_end: u64,
1181) -> Result<(u64, u64), FileError> {
1182 if absolute_offset >= container_end {
1183 return Err(FileError::Container("truncated container varint"));
1184 }
1185 let remaining = container_end - absolute_offset;
1186 let probe_len = remaining.min(10);
1187 let bytes = reader.read_at(absolute_offset, probe_len)?;
1188 read_uvarint(&bytes)
1189 .map(|(value, used)| (value, used as u64))
1190 .ok_or(FileError::Container("truncated container varint"))
1191}
1192
1193fn locate_container_section_ranged<R: RangeReader>(
1196 reader: &R,
1197 container_offset: u64,
1198 container_len: u64,
1199 section_index: usize,
1200 expected_sections: u64,
1201) -> Result<ByteRange, FileError> {
1202 let container_end = checked_end(container_offset, container_len)?;
1203 let (section_count, used) = read_uvarint_at(reader, container_offset, container_end)?;
1204 if section_count != expected_sections {
1205 return Err(FileError::Container("unexpected container section count"));
1206 }
1207 if section_index >= section_count as usize {
1208 return Err(FileError::Container(
1209 "container section index out of bounds",
1210 ));
1211 }
1212
1213 let mut pos = checked_end(container_offset, used)?;
1214 for i in 0..section_count as usize {
1215 let (payload_len, len_used) = read_uvarint_at(reader, pos, container_end)?;
1216 pos = checked_end(pos, len_used)?;
1217 let payload_end = checked_end(pos, payload_len)?;
1218 if payload_end > container_end {
1219 return Err(FileError::Container("section overruns buffer"));
1220 }
1221 if i == section_index {
1222 return Ok(ByteRange {
1223 offset: pos,
1224 len: payload_len,
1225 });
1226 }
1227 pos = payload_end;
1228 }
1229 Err(FileError::Container("container section not found"))
1230}
1231
1232pub fn write_file(
1236 dict: &Dictionary,
1237 index: &GraphIndex,
1238 has_quads: bool,
1239 pyramid_meta: &[u8],
1240 pyramid_levels: u16,
1241) -> Vec<u8> {
1242 write_dataset(dict, index, &[], has_quads, pyramid_meta, pyramid_levels)
1243}
1244
1245fn encode_index_container(index: &GraphIndex, codec: u8) -> Vec<u8> {
1248 let payloads = index
1249 .tile_sections()
1250 .map(|tiles| encode_tiled_section(tiles, codec));
1251 let refs: Vec<&[u8]> = payloads.iter().map(|p| p.as_slice()).collect();
1252 encode_container(&refs, CODEC_NONE)
1253}
1254
1255fn encode_named_graphs(named: &[(String, GraphIndex)], codec: u8) -> Vec<u8> {
1257 let mut out = Vec::new();
1258 write_uvarint(&mut out, named.len() as u64);
1259 for (iri, index) in named {
1260 write_uvarint(&mut out, iri.len() as u64);
1261 out.extend_from_slice(iri.as_bytes());
1262 let container = encode_index_container(index, codec);
1263 write_uvarint(&mut out, container.len() as u64);
1264 out.extend_from_slice(&container);
1265 }
1266 out
1267}
1268
1269fn decode_named_graphs(bytes: &[u8], codec: u8) -> Result<Vec<(String, GraphIndex)>, FileError> {
1270 let (n, mut pos) = read_uvarint(bytes).ok_or(FileError::Container("truncated graph count"))?;
1271 let bound = |start: usize, len: u64| -> Result<usize, FileError> {
1274 start
1275 .checked_add(len as usize)
1276 .filter(|&e| e <= bytes.len())
1277 .ok_or(FileError::Container("named-graph field overruns buffer"))
1278 };
1279 let mut out = Vec::with_capacity((n as usize).min(bytes.len()));
1280 for _ in 0..n {
1281 let (ilen, u1) = read_uvarint(bytes.get(pos..).unwrap_or(&[]))
1282 .ok_or(FileError::Container("truncated iri len"))?;
1283 pos += u1;
1284 let iend = bound(pos, ilen)?;
1285 let iri = String::from_utf8_lossy(&bytes[pos..iend]).into_owned();
1286 pos = iend;
1287 let (clen, u2) = read_uvarint(bytes.get(pos..).unwrap_or(&[]))
1288 .ok_or(FileError::Container("truncated container len"))?;
1289 pos += u2;
1290 let cend = bound(pos, clen)?;
1291 let index = decode_index_container(&bytes[pos..cend], codec)?;
1292 out.push((iri, index));
1293 pos = cend;
1294 }
1295 Ok(out)
1296}
1297
1298pub fn write_dataset(
1301 dict: &Dictionary,
1302 default_index: &GraphIndex,
1303 named: &[(String, GraphIndex)],
1304 has_quads: bool,
1305 pyramid_meta: &[u8],
1306 pyramid_levels: u16,
1307) -> Vec<u8> {
1308 write_dataset_with_metadata(
1309 dict,
1310 default_index,
1311 named,
1312 has_quads,
1313 pyramid_meta,
1314 pyramid_levels,
1315 &[],
1316 &[],
1317 )
1318}
1319
1320pub(crate) fn encode_dict_container(dict: &Dictionary, codec: u8) -> Vec<u8> {
1325 let raw_sections = dict.sections();
1326 let dict_payloads: Vec<Vec<u8>> = raw_sections
1327 .iter()
1328 .map(|raw| encode_chunked_dict_section(raw, codec))
1329 .collect();
1330 encode_container(
1331 &[
1332 dict_payloads[0].as_slice(),
1333 dict_payloads[1].as_slice(),
1334 dict_payloads[2].as_slice(),
1335 dict_payloads[3].as_slice(),
1336 ],
1337 CODEC_NONE,
1338 )
1339}
1340
1341#[allow(clippy::too_many_arguments)]
1352pub fn write_dataset_with_metadata(
1353 dict: &Dictionary,
1354 default_index: &GraphIndex,
1355 named: &[(String, GraphIndex)],
1356 has_quads: bool,
1357 pyramid_meta: &[u8],
1358 pyramid_levels: u16,
1359 metadata: &[u8],
1360 text_index: &[u8],
1361) -> Vec<u8> {
1362 let codec = writer_codec();
1363 let dict_container = encode_dict_container(dict, codec);
1364 write_dataset_from_parts(
1365 &dict_container,
1366 dict.term_count() as u64,
1367 default_index,
1368 named,
1369 has_quads,
1370 dict.has_quoted_triples(),
1371 pyramid_meta,
1372 pyramid_levels,
1373 metadata,
1374 text_index,
1375 codec,
1376 )
1377}
1378
1379#[allow(clippy::too_many_arguments)]
1384pub(crate) fn write_dataset_from_parts(
1385 dict_container: &[u8],
1386 term_count: u64,
1387 default_index: &GraphIndex,
1388 named: &[(String, GraphIndex)],
1389 has_quads: bool,
1390 has_quoted_triples: bool,
1391 pyramid_meta: &[u8],
1392 pyramid_levels: u16,
1393 metadata: &[u8],
1394 text_index: &[u8],
1395 codec: u8,
1396) -> Vec<u8> {
1397 let index_container = encode_index_container(default_index, codec);
1398 let named_section = encode_named_graphs(named, codec);
1399
1400 let meta_section_len = metadata.len() as u64;
1403 let dict_offset = HEADER_LEN as u64 + meta_section_len;
1404 let dict_len = dict_container.len() as u64;
1405 let index_offset = dict_offset + dict_len;
1406 let index_len = index_container.len() as u64;
1407 let pyr_offset = index_offset + index_len;
1408 let pyr_len = pyramid_meta.len() as u64;
1409 let text_offset = pyr_offset + pyr_len;
1411 let text_len = text_index.len() as u64;
1412 let named_offset = text_offset + text_len;
1413 let named_len = if named.is_empty() {
1414 0
1415 } else {
1416 named_section.len() as u64
1417 };
1418
1419 let mut parts: Vec<&[u8]> = Vec::with_capacity(5);
1425 if meta_section_len > 0 {
1426 parts.push(metadata);
1427 }
1428 parts.push(dict_container);
1429 parts.push(&index_container);
1430 parts.push(pyramid_meta);
1431 if text_len > 0 {
1432 parts.push(text_index);
1433 }
1434 if named_len > 0 {
1435 parts.push(&named_section);
1436 }
1437
1438 let schema_meta_len = crate::meta::schema_block_len(pyramid_meta);
1441
1442 let header = Header {
1443 version: crate::header::CURRENT_FORMAT_VERSION,
1444 flags: FLAG_TILE_SYNOPSIS
1445 | if has_quads { FLAG_HAS_QUADS } else { 0 }
1446 | if has_quoted_triples {
1447 FLAG_HAS_QUOTED_TRIPLES
1448 } else {
1449 0
1450 },
1451 metadata_offset: HEADER_LEN as u64,
1452 metadata_len: meta_section_len,
1453 dictionary_offset: dict_offset,
1454 dictionary_len: dict_len,
1455 root_dir_offset: index_offset,
1456 root_dir_len: index_len,
1457 pyramid_meta_offset: if pyr_len > 0 { pyr_offset } else { 0 },
1458 pyramid_meta_len: pyr_len,
1459 dict_codec: codec,
1460 block_codec: codec,
1461 pyramid_levels,
1462 quad_count: default_index.triple_count() as u64
1463 + named
1464 .iter()
1465 .map(|(_, idx)| idx.triple_count() as u64)
1466 .sum::<u64>(),
1467 term_count,
1468 content_hash: content_hash(&parts),
1469 named_graphs_offset: if named_len > 0 { named_offset } else { 0 },
1470 named_graphs_len: named_len,
1471 schema_meta_len,
1472 text_index_offset: if text_len > 0 { text_offset } else { 0 },
1473 text_index_len: text_len,
1474 extra_sections: Vec::new(),
1475 };
1476
1477 let mut out = Vec::with_capacity(
1478 HEADER_LEN
1479 + metadata.len()
1480 + dict_container.len()
1481 + index_container.len()
1482 + pyramid_meta.len()
1483 + text_index.len()
1484 + named_section.len()
1485 + MAGIC.len(),
1486 );
1487 out.extend_from_slice(&header.to_bytes());
1488 if meta_section_len > 0 {
1489 out.extend_from_slice(metadata);
1490 }
1491 out.extend_from_slice(dict_container);
1492 out.extend_from_slice(&index_container);
1493 out.extend_from_slice(pyramid_meta);
1494 if text_len > 0 {
1495 out.extend_from_slice(text_index);
1496 }
1497 if named_len > 0 {
1498 out.extend_from_slice(&named_section);
1499 }
1500 out.extend_from_slice(&MAGIC); out
1502}
1503
1504pub const RDF_TYPE: &str = "<http://www.w3.org/1999/02/22-rdf-syntax-ns#type>";
1506
1507pub fn schema_summary(rete: &Rete) -> Vec<(String, String, String, u32)> {
1514 use std::collections::{BTreeMap, HashMap};
1515 let triples = rete.dump(None);
1516
1517 let mut class_of: HashMap<&str, &str> = HashMap::new();
1518 for (s, p, o) in &triples {
1519 if p == RDF_TYPE {
1520 class_of.insert(s.as_str(), o.as_str());
1521 }
1522 }
1523 let classify = |t: &str| -> String {
1524 if let Some(c) = class_of.get(t) {
1525 (*c).to_string()
1526 } else if t.starts_with('"') {
1527 "(literal)".to_string()
1528 } else {
1529 "(untyped)".to_string()
1530 }
1531 };
1532
1533 let mut counts: BTreeMap<(String, String, String), u32> = BTreeMap::new();
1534 for (s, p, o) in &triples {
1535 if p == RDF_TYPE {
1536 continue; }
1538 *counts
1539 .entry((classify(s), p.clone(), classify(o)))
1540 .or_default() += 1;
1541 }
1542 counts
1543 .into_iter()
1544 .map(|((a, p, b), c)| (a, p, b, c))
1545 .collect()
1546}
1547
1548pub fn schema_classes(rete: &Rete) -> Vec<(String, u32)> {
1552 use std::collections::BTreeMap;
1553 let mut counts: BTreeMap<String, u32> = BTreeMap::new();
1554 for (_s, p, o) in rete.dump(None) {
1555 if p == RDF_TYPE {
1556 *counts.entry(o).or_default() += 1;
1557 }
1558 }
1559 let mut out: Vec<(String, u32)> = counts.into_iter().collect();
1560 out.sort_by(|a, b| b.1.cmp(&a.1).then_with(|| a.0.cmp(&b.0)));
1561 out
1562}
1563
1564pub fn read_metadata_ranged<R: RangeReader>(reader: &R) -> Result<Option<Vec<u8>>, FileError> {
1574 let head = reader.read_at(0, HEADER_LEN as u64)?;
1575 let header = Header::from_bytes(&head)?;
1576 if header.metadata_len == 0 {
1577 return Ok(None);
1578 }
1579 let bytes = reader.read_at(header.metadata_offset, header.metadata_len)?;
1580 Ok(Some(bytes))
1581}
1582
1583pub fn verify(bytes: &[u8]) -> Result<bool, FileError> {
1586 let header = Header::from_bytes(bytes)?;
1587 let slice = |off: u64, len: u64| -> Result<&[u8], FileError> {
1588 bytes
1589 .get(off as usize..(off + len) as usize)
1590 .ok_or(FileError::Container("section overruns buffer"))
1591 };
1592 let d = slice(header.dictionary_offset, header.dictionary_len)?;
1593 let i = slice(header.root_dir_offset, header.root_dir_len)?;
1594 let m = if header.pyramid_meta_len > 0 {
1595 slice(header.pyramid_meta_offset, header.pyramid_meta_len)?
1596 } else {
1597 &[]
1598 };
1599 let mut parts: Vec<&[u8]> = Vec::with_capacity(6);
1603 if header.metadata_len > 0 {
1604 parts.push(slice(header.metadata_offset, header.metadata_len)?);
1605 }
1606 parts.push(d);
1607 parts.push(i);
1608 parts.push(m);
1609 if header.text_index_len > 0 {
1610 parts.push(slice(header.text_index_offset, header.text_index_len)?);
1611 }
1612 if header.named_graphs_len > 0 {
1613 parts.push(slice(header.named_graphs_offset, header.named_graphs_len)?);
1614 }
1615 Ok(content_hash(&parts) == header.content_hash)
1616}
1617
1618type PyramidLoader = Box<dyn Fn() -> Option<PyramidMeta> + Send + Sync>;
1620
1621enum PyramidSlot {
1627 Resident(Option<PyramidMeta>),
1628 Lazy {
1629 loader: PyramidLoader,
1630 cell: std::sync::OnceLock<Option<PyramidMeta>>,
1631 },
1632}
1633
1634type TextIndexLoader = Box<dyn Fn() -> Option<crate::text_index::TextIndex> + Send + Sync>;
1637
1638enum TextIndexSlot {
1643 Resident(Option<crate::text_index::TextIndex>),
1644 Lazy {
1645 loader: TextIndexLoader,
1646 cell: std::sync::OnceLock<Option<crate::text_index::TextIndex>>,
1647 },
1648}
1649
1650pub struct Rete {
1652 header: Header,
1653 dict: Dictionary,
1654 index: GraphIndex,
1655 index_section_ranges: [ByteRange; NUM_PERMS],
1656 tile_ranges: [Vec<(u32, u32, ByteRange)>; NUM_PERMS],
1660 pyramid: PyramidSlot,
1661 text_index: TextIndexSlot,
1662 named_graphs: Vec<(String, GraphIndex)>,
1663 metadata: Vec<u8>,
1668 service_client: Option<Box<dyn crate::service::ServiceClient>>,
1673 service_error: std::sync::Mutex<Option<String>>,
1677}
1678
1679impl Rete {
1680 pub fn open(bytes: &[u8]) -> Result<Self, FileError> {
1683 let header = Header::from_bytes(bytes)?;
1684
1685 let region = |off: u64, len: u64| -> Result<&[u8], FileError> {
1689 let start = off as usize;
1690 let end = start
1691 .checked_add(len as usize)
1692 .filter(|&e| e <= bytes.len())
1693 .ok_or(FileError::Container("section range out of bounds"))?;
1694 Ok(&bytes[start..end])
1695 };
1696
1697 let dict = decode_dictionary_container(
1698 region(header.dictionary_offset, header.dictionary_len)?,
1699 header.dict_codec,
1700 )?;
1701
1702 let index_bytes = region(header.root_dir_offset, header.root_dir_len)?;
1703 let index = decode_index_container(index_bytes, header.block_codec)?;
1704 let index_section_ranges =
1705 decode_index_section_ranges(index_bytes, header.root_dir_offset)?;
1706
1707 let pyramid = PyramidSlot::Resident(if header.pyramid_meta_len > 0 {
1708 Some(
1709 PyramidMeta::decode(region(header.pyramid_meta_offset, header.pyramid_meta_len)?)
1710 .map_err(|_| FileError::Container("malformed pyramid meta"))?,
1711 )
1712 } else {
1713 None
1714 });
1715
1716 let text_index = TextIndexSlot::Resident(if header.text_index_len > 0 {
1719 Some(
1720 crate::text_index::TextIndex::from_section(
1721 region(header.text_index_offset, header.text_index_len)?,
1722 header.block_codec,
1723 )
1724 .map_err(|_| FileError::Container("malformed text index"))?,
1725 )
1726 } else {
1727 None
1728 });
1729
1730 let named_graphs = if header.named_graphs_len > 0 {
1731 decode_named_graphs(
1732 region(header.named_graphs_offset, header.named_graphs_len)?,
1733 header.block_codec,
1734 )?
1735 } else {
1736 Vec::new()
1737 };
1738
1739 let metadata = if header.metadata_len > 0 {
1740 region(header.metadata_offset, header.metadata_len)?.to_vec()
1741 } else {
1742 Vec::new()
1743 };
1744
1745 let tile_ranges =
1746 tile_file_ranges(index_bytes, header.root_dir_offset, &index_section_ranges);
1747 Ok(Self {
1748 header,
1749 dict,
1750 index,
1751 index_section_ranges,
1752 tile_ranges,
1753 pyramid,
1754 text_index,
1755 named_graphs,
1756 metadata,
1757 service_client: None,
1758 service_error: std::sync::Mutex::new(None),
1759 })
1760 }
1761
1762 pub fn set_service_client(&mut self, client: Box<dyn crate::service::ServiceClient>) {
1767 self.service_client = Some(client);
1768 }
1769
1770 pub(crate) fn service_client(&self) -> Option<&dyn crate::service::ServiceClient> {
1771 self.service_client.as_deref()
1772 }
1773
1774 pub(crate) fn record_service_error(&self, msg: &str) {
1776 let mut e = self.service_error.lock().unwrap();
1777 if e.is_none() {
1778 *e = Some(msg.to_string());
1779 }
1780 }
1781
1782 pub(crate) fn take_service_error(&self) -> Option<String> {
1785 self.service_error.lock().unwrap().take()
1786 }
1787
1788 pub fn header(&self) -> &Header {
1789 &self.header
1790 }
1791
1792 pub fn file_layout(&self) -> Vec<LayoutSegment> {
1798 let h = &self.header;
1799 let seg = |kind: &'static str, label: String, offset: u64, len: u64| LayoutSegment {
1800 kind,
1801 label,
1802 offset,
1803 len,
1804 };
1805 let mut out = vec![seg(
1806 "header",
1807 "header (fixed 128 bytes)".into(),
1808 0,
1809 crate::header::HEADER_LEN as u64,
1810 )];
1811 if h.metadata_len > 0 {
1812 out.push(seg(
1813 "metadata",
1814 "metadata (dataset card)".into(),
1815 h.metadata_offset,
1816 h.metadata_len,
1817 ));
1818 }
1819 out.push(seg(
1820 "dictionary",
1821 "dictionary (4 front-coded term sections)".into(),
1822 h.dictionary_offset,
1823 h.dictionary_len,
1824 ));
1825 for (si, perm) in crate::index::ALL_PERMS.into_iter().enumerate() {
1826 let sec = self.index_section_ranges[si];
1827 if sec.len == 0 {
1828 continue;
1829 }
1830 let first_tile = self.tile_ranges[si]
1831 .first()
1832 .map(|&(_, _, r)| r.offset)
1833 .unwrap_or(sec.offset + sec.len);
1834 if first_tile > sec.offset {
1835 out.push(seg(
1836 "directory",
1837 format!("{} tile directory", perm.name()),
1838 sec.offset,
1839 first_tile - sec.offset,
1840 ));
1841 }
1842 for (ti, &(min_a, max_a, r)) in self.tile_ranges[si].iter().enumerate() {
1843 out.push(seg(
1844 "tile",
1845 format!("{} tile {ti} (leading ids {min_a}..{max_a})", perm.name()),
1846 r.offset,
1847 r.len,
1848 ));
1849 }
1850 }
1851 if h.pyramid_meta_len > 0 {
1852 out.push(seg(
1853 "pyramid",
1854 "pyramid summary (communities + superedges)".into(),
1855 h.pyramid_meta_offset,
1856 h.pyramid_meta_len,
1857 ));
1858 }
1859 if h.named_graphs_len > 0 {
1860 out.push(seg(
1861 "named-graphs",
1862 format!("named graphs ({})", self.named_graphs.len()),
1863 h.named_graphs_offset,
1864 h.named_graphs_len,
1865 ));
1866 }
1867 out.sort_by_key(|s| s.offset);
1868 out
1869 }
1870
1871 pub fn metadata(&self) -> Option<&[u8]> {
1876 if self.metadata.is_empty() {
1877 None
1878 } else {
1879 Some(&self.metadata)
1880 }
1881 }
1882
1883 pub fn dictionary(&self) -> &Dictionary {
1884 &self.dict
1885 }
1886
1887 pub fn pyramid(&self) -> Option<&PyramidMeta> {
1889 match &self.pyramid {
1890 PyramidSlot::Resident(p) => p.as_ref(),
1891 PyramidSlot::Lazy { loader, cell } => cell.get_or_init(loader).as_ref(),
1893 }
1894 }
1895
1896 pub fn pyramid_if_loaded(&self) -> Option<&PyramidMeta> {
1901 match &self.pyramid {
1902 PyramidSlot::Resident(p) => p.as_ref(),
1903 PyramidSlot::Lazy { cell, .. } => cell.get().and_then(|o| o.as_ref()),
1904 }
1905 }
1906
1907 pub fn predicate_stats(&self) -> &[crate::meta::PredStat] {
1911 self.pyramid_if_loaded()
1912 .map(|p| p.predicate_stats.as_slice())
1913 .unwrap_or(&[])
1914 }
1915
1916 pub fn char_sets(&self) -> &[crate::meta::CharSet] {
1919 self.pyramid_if_loaded()
1920 .map(|p| p.char_sets.as_slice())
1921 .unwrap_or(&[])
1922 }
1923
1924 pub fn label_index(&self) -> &[crate::meta::LabelEntry] {
1927 self.pyramid_if_loaded()
1928 .map(|p| p.label_index.as_slice())
1929 .unwrap_or(&[])
1930 }
1931
1932 pub fn prefix_search(&self, prefix: &str, limit: usize) -> Vec<(String, String)> {
1938 let Some(pyr) = self.pyramid() else {
1939 return Vec::new();
1940 };
1941 pyr.prefix_search(prefix, limit)
1942 .into_iter()
1943 .filter_map(|e| {
1944 self.dict
1945 .subject_term(e.subject)
1946 .map(|iri| (e.label.clone(), iri))
1947 })
1948 .collect()
1949 }
1950
1951 pub(crate) fn text_index(&self) -> Option<&crate::text_index::TextIndex> {
1954 match &self.text_index {
1955 TextIndexSlot::Resident(t) => t.as_ref(),
1956 TextIndexSlot::Lazy { loader, cell } => cell.get_or_init(loader).as_ref(),
1957 }
1958 }
1959
1960 pub fn has_text_index(&self) -> bool {
1963 self.header.text_index_len > 0
1964 }
1965
1966 pub fn text_search(&self, words: &[&str], prefix: Option<&str>, limit: usize) -> Vec<String> {
1976 let Some(ti) = self.text_index() else {
1977 return Vec::new();
1978 };
1979 let mut acc: Option<Vec<u32>> = None;
1983 if let Some(p) = prefix {
1984 acc = Some(ti.prefix(&p.to_lowercase()));
1985 }
1986 for w in words {
1987 for tok in crate::text_index::tokenize(w) {
1988 let posting = ti.lookup(&tok);
1989 acc = Some(match acc {
1990 Some(a) => intersect_sorted(&a, &posting),
1991 None => posting,
1992 });
1993 if acc.as_ref().is_some_and(|a| a.is_empty()) {
1994 return Vec::new();
1995 }
1996 }
1997 }
1998 let ids = acc.unwrap_or_default();
1999 let mut out = Vec::with_capacity(if limit > 0 {
2000 limit.min(ids.len())
2001 } else {
2002 ids.len()
2003 });
2004 for id in ids {
2005 if let Some(iri) = self.dict.subject_term(id) {
2006 out.push(iri);
2007 if limit > 0 && out.len() >= limit {
2008 break;
2009 }
2010 }
2011 }
2012 out
2013 }
2014
2015 pub fn default_index(&self) -> &GraphIndex {
2017 &self.index
2018 }
2019
2020 pub fn dump(&self, graph: Option<&str>) -> Vec<TermTriple> {
2022 self.dict.prefetch_all();
2025 let index = match graph {
2026 None => &self.index,
2027 Some(g) => match self.graph_index(g) {
2028 Some(i) => i,
2029 None => return Vec::new(),
2030 },
2031 };
2032 index
2033 .match_pattern((None, None, None))
2034 .into_iter()
2035 .filter_map(|(s, p, o)| {
2036 Some((
2037 self.dict.subject_term(s)?,
2038 self.dict.predicate_term(p)?,
2039 self.dict.object_term(o)?,
2040 ))
2041 })
2042 .collect()
2043 }
2044
2045 pub fn dump_each<F: FnMut(&str, &str, &str)>(&self, graph: Option<&str>, mut f: F) {
2050 self.dict.prefetch_all();
2051 let index = match graph {
2052 None => &self.index,
2053 Some(g) => match self.graph_index(g) {
2054 Some(i) => i,
2055 None => return,
2056 },
2057 };
2058 for (s, p, o) in index.scan_iter((None, None, None)) {
2059 if let (Some(st), Some(pt), Some(ot)) = (
2060 self.dict.subject_term(s),
2061 self.dict.predicate_term(p),
2062 self.dict.object_term(o),
2063 ) {
2064 f(&st, &pt, &ot);
2065 }
2066 }
2067 }
2068
2069 pub fn named_graphs(&self) -> &[(String, GraphIndex)] {
2071 &self.named_graphs
2072 }
2073
2074 pub fn graph_names(&self) -> Vec<&str> {
2076 self.named_graphs
2077 .iter()
2078 .map(|(iri, _)| iri.as_str())
2079 .collect()
2080 }
2081
2082 pub fn graph_index(&self, iri: &str) -> Option<&GraphIndex> {
2084 self.named_graphs
2085 .iter()
2086 .find(|(name, _)| name == iri)
2087 .map(|(_, idx)| idx)
2088 }
2089
2090 pub fn match_ids(
2093 &self,
2094 pattern: (Option<u32>, Option<u32>, Option<u32>),
2095 ) -> Vec<(u32, u32, u32)> {
2096 self.index.match_pattern(pattern)
2097 }
2098
2099 pub fn predicate_pairs(&self, predicate: &str) -> Vec<(u32, u32)> {
2102 let pid = match self.dict.predicate_id(predicate) {
2103 Some(p) => p,
2104 None => return Vec::new(),
2105 };
2106 self.index
2107 .match_pattern((None, Some(pid), None))
2108 .into_iter()
2109 .map(|(s, _p, o)| (self.dict.subject_node(s), self.dict.object_node(o)))
2110 .collect()
2111 }
2112
2113 pub fn open_ranged<R: RangeReader>(reader: &R) -> Result<Self, FileError> {
2117 let head = reader.read_at(0, HEADER_LEN as u64)?;
2118 let header = Header::from_bytes(&head)?;
2119
2120 let dict_bytes = reader.read_at(header.dictionary_offset, header.dictionary_len)?;
2121 let dict = decode_dictionary_container(&dict_bytes, header.dict_codec)?;
2122
2123 let index_bytes = reader.read_at(header.root_dir_offset, header.root_dir_len)?;
2124 let index = decode_index_container(&index_bytes, header.block_codec)?;
2125 let index_section_ranges =
2126 decode_index_section_ranges(&index_bytes, header.root_dir_offset)?;
2127
2128 let pyramid = PyramidSlot::Resident(if header.pyramid_meta_len > 0 {
2129 let mb = reader.read_at(header.pyramid_meta_offset, header.pyramid_meta_len)?;
2130 Some(
2131 PyramidMeta::decode(&mb)
2132 .map_err(|_| FileError::Container("malformed pyramid meta"))?,
2133 )
2134 } else {
2135 None
2136 });
2137
2138 let text_index = TextIndexSlot::Resident(if header.text_index_len > 0 {
2141 let tb = reader.read_at(header.text_index_offset, header.text_index_len)?;
2142 Some(
2143 crate::text_index::TextIndex::from_section(&tb, header.block_codec)
2144 .map_err(|_| FileError::Container("malformed text index"))?,
2145 )
2146 } else {
2147 None
2148 });
2149
2150 let named_graphs = if header.named_graphs_len > 0 {
2151 let nb = reader.read_at(header.named_graphs_offset, header.named_graphs_len)?;
2152 decode_named_graphs(&nb, header.block_codec)?
2153 } else {
2154 Vec::new()
2155 };
2156
2157 let tile_ranges =
2161 tile_file_ranges(&index_bytes, header.root_dir_offset, &index_section_ranges);
2162 Ok(Self {
2163 header,
2164 dict,
2165 index,
2166 index_section_ranges,
2167 tile_ranges,
2168 pyramid,
2169 text_index,
2170 named_graphs,
2171 metadata: Vec::new(),
2172 service_client: None,
2173 service_error: std::sync::Mutex::new(None),
2174 })
2175 }
2176
2177 pub fn open_ranged_lazy<R: RangeReader + Send + Sync + 'static>(
2189 reader: R,
2190 ) -> Result<Self, FileError> {
2191 let head = reader.read_at(0, HEADER_LEN as u64)?;
2192 let header = Header::from_bytes(&head)?;
2193 let reader = std::sync::Arc::new(reader);
2194 let read_concurrency = reader.concurrency();
2197
2198 let mut dict_sections: Vec<crate::dict::ChunkedSection> = Vec::with_capacity(4);
2202 for si in 0..4 {
2203 let section = locate_container_section_ranged(
2204 reader.as_ref(),
2205 header.dictionary_offset,
2206 header.dictionary_len,
2207 si,
2208 4,
2209 )?;
2210 let (meta, entries) = read_dict_dir_ranged(reader.as_ref(), section)?;
2211 let ranges: Vec<ByteRange> = entries
2212 .iter()
2213 .map(|e| ByteRange {
2214 offset: section.offset + e.start,
2215 len: (e.end - e.start),
2216 })
2217 .collect();
2218 let chunks: Vec<crate::dict::SectionChunk> = entries
2219 .into_iter()
2220 .map(|e| crate::dict::SectionChunk::remote(e.first_run, e.first_term, e.body_start))
2221 .collect();
2222 let chunk_reader = reader.clone();
2223 let codec = header.dict_codec;
2224 let loader_ranges = ranges.clone();
2225 let loader: crate::dict::ChunkLoader = Box::new(move |ci| {
2226 let range = loader_ranges.get(ci)?;
2227 let bytes = chunk_reader.read_at(range.offset, range.len).ok()?;
2228 decompress(codec, &bytes).ok()
2229 });
2230 let bulk_reader = reader.clone();
2233 let bulk: crate::dict::ChunkBulkLoader = Box::new(move |cis| {
2234 let want: Option<Vec<ByteRange>> =
2235 cis.iter().map(|&ci| ranges.get(ci).copied()).collect();
2236 let blobs = read_coalesced(bulk_reader.as_ref(), &want?, DICT_COALESCE_GAP)?;
2237 blobs.iter().map(|b| decompress(codec, b).ok()).collect()
2238 });
2239 dict_sections.push(
2240 crate::dict::ChunkedSection::from_parts(meta, chunks, Some(loader))
2241 .with_bulk_loader(bulk),
2242 );
2243 }
2244 let dict_arr: [crate::dict::ChunkedSection; 4] = dict_sections
2245 .try_into()
2246 .map_err(|_| FileError::Container("expected 4 dictionary sections"))?;
2247 let dict = Dictionary::from_chunked_sections(dict_arr);
2248
2249 let mut index_section_ranges = [ByteRange { offset: 0, len: 0 }; NUM_PERMS];
2252 let mut tile_ranges: [Vec<(u32, u32, ByteRange)>; NUM_PERMS] = Default::default();
2253 #[allow(clippy::type_complexity)]
2254 let mut directories: [Vec<(u32, u32, Option<TileSynopsis>)>; NUM_PERMS] =
2255 Default::default();
2256 for si in 0..NUM_PERMS {
2257 let section = locate_container_section_ranged(
2258 reader.as_ref(),
2259 header.root_dir_offset,
2260 header.root_dir_len,
2261 si,
2262 NUM_PERMS as u64,
2263 )?;
2264 index_section_ranges[si] = section;
2265 let dir = read_tile_directory_ranged(reader.as_ref(), section)?;
2266 let syn = if header.has_tile_synopsis() {
2269 read_tile_synopsis_ranged(reader.as_ref(), section, &dir)
2270 } else {
2271 vec![None; dir.len()]
2272 };
2273 directories[si] = dir
2274 .iter()
2275 .zip(syn)
2276 .map(|(e, s)| (e.min_a, e.max_a, s))
2277 .collect();
2278 tile_ranges[si] = dir
2279 .into_iter()
2280 .map(|e| {
2281 (
2282 e.min_a,
2283 e.max_a,
2284 ByteRange {
2285 offset: section.offset + e.start,
2286 len: (e.end - e.start),
2287 },
2288 )
2289 })
2290 .collect();
2291 }
2292
2293 let pyramid = if header.pyramid_meta_len > 0 {
2297 let pyr_reader = reader.clone();
2298 let pyr_off = header.pyramid_meta_offset;
2299 let pyr_len = header.pyramid_meta_len;
2300 PyramidSlot::Lazy {
2301 loader: Box::new(move || {
2302 let mb = pyr_reader.read_at(pyr_off, pyr_len).ok()?;
2303 PyramidMeta::decode(&mb).ok()
2304 }),
2305 cell: std::sync::OnceLock::new(),
2306 }
2307 } else {
2308 PyramidSlot::Resident(None)
2309 };
2310
2311 let text_index = if header.text_index_len > 0 {
2316 let ti_reader = reader.clone();
2317 let ti_off = header.text_index_offset;
2318 let ti_len = header.text_index_len;
2319 let codec = header.block_codec;
2320 TextIndexSlot::Lazy {
2321 loader: Box::new(move || {
2322 let head_len = 10u64.min(ti_len);
2326 let head = ti_reader.read_at(ti_off, head_len).ok()?;
2327 let (ttlen, n) = crate::varint::read_uvarint(&head)?;
2328 let prefix_len = (n as u64 + ttlen).min(ti_len);
2329 let prefix = ti_reader.read_at(ti_off, prefix_len).ok()?;
2330 let postings_base =
2331 crate::text_index::TextIndex::postings_base(&prefix)? as u64;
2332 let postings_abs = ti_off + postings_base;
2333 let pr = ti_reader.clone();
2334 let posting_loader = Box::new(move |off: u64, len: u64| {
2335 pr.read_at(postings_abs + off, len).ok()
2336 });
2337 crate::text_index::TextIndex::from_token_table(&prefix, codec, posting_loader)
2338 .ok()
2339 }),
2340 cell: std::sync::OnceLock::new(),
2341 }
2342 } else {
2343 TextIndexSlot::Resident(None)
2344 };
2345
2346 let named_graphs = if header.named_graphs_len > 0 {
2347 let nb = reader.read_at(header.named_graphs_offset, header.named_graphs_len)?;
2348 decode_named_graphs(&nb, header.block_codec)?
2349 } else {
2350 Vec::new()
2351 };
2352
2353 let codec = header.block_codec;
2358 let loader_ranges = tile_ranges.clone();
2359 let loader_reader = reader.clone();
2360 let loader: crate::index::TileLoader = Box::new(move |si, ti| {
2361 let (_, _, range) = loader_ranges.get(si)?.get(ti)?;
2362 let bytes = loader_reader.read_at(range.offset, range.len).ok()?;
2363 decompress(codec, &bytes).ok()
2364 });
2365 let bulk_ranges = tile_ranges.clone();
2366 let bulk: crate::index::TileBulkLoader = Box::new(move |si, tis| {
2367 let section = bulk_ranges.get(si)?;
2368 let want: Option<Vec<ByteRange>> = tis
2369 .iter()
2370 .map(|&ti| section.get(ti).map(|&(_, _, r)| r))
2371 .collect();
2372 let blobs = read_coalesced(reader.as_ref(), &want?, TILE_COALESCE_GAP)?;
2373 blobs.iter().map(|b| decompress(codec, b).ok()).collect()
2374 });
2375 let mut index =
2376 GraphIndex::from_remote_directories(directories, loader).with_bulk_loader(bulk);
2377 index.set_tile_lens(std::array::from_fn(|si| {
2380 tile_ranges[si]
2381 .iter()
2382 .map(|&(_, _, r)| r.len.min(u32::MAX as u64) as u32)
2383 .collect()
2384 }));
2385 index.set_read_concurrency(read_concurrency);
2389
2390 Ok(Self {
2391 header,
2392 dict,
2393 index,
2394 index_section_ranges,
2395 tile_ranges,
2396 pyramid,
2397 text_index,
2398 named_graphs,
2399 metadata: Vec::new(),
2400 service_client: None,
2401 service_error: std::sync::Mutex::new(None),
2402 })
2403 }
2404
2405 pub fn index_incomplete(&self) -> bool {
2410 self.index.load_incomplete()
2411 || self.dict.load_incomplete()
2412 || self.named_graphs.iter().any(|(_, g)| g.load_incomplete())
2413 }
2414
2415 pub fn reset_load_failures(&self) {
2423 self.index.reset_load_failure();
2424 self.dict.reset_load_failure();
2425 for (_, g) in &self.named_graphs {
2426 g.reset_load_failure();
2427 }
2428 }
2429
2430 fn resolve_query_pattern(
2431 &self,
2432 s: Option<&str>,
2433 p: Option<&str>,
2434 o: Option<&str>,
2435 ) -> Option<Pattern> {
2436 let sid = match s {
2437 Some(t) => match self.dict.subject_id(t) {
2438 Some(id) => Some(id),
2439 None => return None,
2440 },
2441 None => None,
2442 };
2443 let pid = match p {
2444 Some(t) => match self.dict.predicate_id(t) {
2445 Some(id) => Some(id),
2446 None => return None,
2447 },
2448 None => None,
2449 };
2450 let oid = match o {
2451 Some(t) => match self.dict.object_id(t) {
2452 Some(id) => Some(id),
2453 None => return None,
2454 },
2455 None => None,
2456 };
2457 Some((sid, pid, oid))
2458 }
2459
2460 pub fn query_with_provenance(
2464 &self,
2465 s: Option<&str>,
2466 p: Option<&str>,
2467 o: Option<&str>,
2468 ) -> Vec<TripleProvenance> {
2469 let pattern = match self.resolve_query_pattern(s, p, o) {
2470 Some(pattern) => pattern,
2471 None => return Vec::new(),
2472 };
2473
2474 let index_permutation = GraphIndex::best_permutation(pattern);
2475 let dictionary_range = ByteRange {
2476 offset: self.header.dictionary_offset,
2477 len: self.header.dictionary_len,
2478 };
2479 let index_range = ByteRange {
2480 offset: self.header.root_dir_offset,
2481 len: self.header.root_dir_len,
2482 };
2483 let index_section_range = self.index_section_ranges[index_permutation.section_index()];
2484 let pyramid_range = (self.header.pyramid_meta_len > 0).then_some(ByteRange {
2485 offset: self.header.pyramid_meta_offset,
2486 len: self.header.pyramid_meta_len,
2487 });
2488
2489 let tiles = &self.tile_ranges[index_permutation.section_index()];
2490 self.index
2491 .match_pattern(pattern)
2492 .into_iter()
2493 .filter_map(|(s, p, o)| {
2494 let terms = (
2495 self.dict.subject_term(s)?,
2496 self.dict.predicate_term(p)?,
2497 self.dict.object_term(o)?,
2498 );
2499 let a = index_permutation.forward((s, p, o)).0;
2502 let ti = tiles.partition_point(|&(_, max_a, _)| max_a < a);
2503 let (tile, tile_range) = match tiles.get(ti) {
2504 Some(&(min_a, _, range)) if min_a <= a => (
2505 Some(format!("{}/{ti}", index_permutation.name())),
2506 Some(range),
2507 ),
2508 _ => (None, None),
2509 };
2510 Some(TripleProvenance {
2511 terms,
2512 ids: (s, p, o),
2513 graph: None,
2514 matched_pattern: pattern,
2515 index_permutation,
2516 dictionary_range,
2517 index_range,
2518 index_section_range,
2519 pyramid_range,
2520 tile,
2521 tile_range,
2522 })
2523 })
2524 .collect()
2525 }
2526
2527 pub fn query(&self, s: Option<&str>, p: Option<&str>, o: Option<&str>) -> Vec<TermTriple> {
2531 self.query_with_provenance(s, p, o)
2532 .into_iter()
2533 .map(|m| m.terms)
2534 .collect()
2535 }
2536
2537 pub fn query_in_graph(
2545 &self,
2546 graph: Option<&str>,
2547 s: Option<&str>,
2548 p: Option<&str>,
2549 o: Option<&str>,
2550 ) -> Vec<TermTriple> {
2551 let pattern = match self.resolve_query_pattern(s, p, o) {
2552 Some(pattern) => pattern,
2553 None => return Vec::new(),
2554 };
2555 let index = match graph {
2556 None => &self.index,
2557 Some(g) => match self.graph_index(g) {
2558 Some(i) => i,
2559 None => return Vec::new(),
2560 },
2561 };
2562 self.dict.prefetch_all();
2563 index
2564 .match_pattern(pattern)
2565 .into_iter()
2566 .filter_map(|(s, p, o)| {
2567 Some((
2568 self.dict.subject_term(s)?,
2569 self.dict.predicate_term(p)?,
2570 self.dict.object_term(o)?,
2571 ))
2572 })
2573 .collect()
2574 }
2575
2576 pub fn query_quads(
2581 &self,
2582 s: Option<&str>,
2583 p: Option<&str>,
2584 o: Option<&str>,
2585 ) -> Vec<(TermTriple, Option<String>)> {
2586 let mut out: Vec<(TermTriple, Option<String>)> = self
2587 .query_in_graph(None, s, p, o)
2588 .into_iter()
2589 .map(|t| (t, None))
2590 .collect();
2591 for (iri, _) in &self.named_graphs {
2592 for triple in self.query_in_graph(Some(iri), s, p, o) {
2593 out.push((triple, Some(iri.clone())));
2594 }
2595 }
2596 out
2597 }
2598
2599 pub fn query_ranged<R: RangeReader>(
2606 reader: &R,
2607 s: Option<&str>,
2608 p: Option<&str>,
2609 o: Option<&str>,
2610 ) -> Result<Vec<TermTriple>, FileError> {
2611 let routed = match route_pattern(reader, s, p, o)? {
2612 Some(routed) => routed,
2613 None => return Ok(Vec::new()),
2614 };
2615 let matches = fetch_routed_matches(reader, &routed)?;
2616 Ok(matches
2617 .into_iter()
2618 .filter_map(|(s, p, o)| {
2619 Some((
2620 routed.dict.subject_term(s)?,
2621 routed.dict.predicate_term(p)?,
2622 routed.dict.object_term(o)?,
2623 ))
2624 })
2625 .collect())
2626 }
2627
2628 pub fn route_pattern_ranged<R: RangeReader>(
2632 reader: &R,
2633 s: Option<&str>,
2634 p: Option<&str>,
2635 o: Option<&str>,
2636 ) -> Result<bool, FileError> {
2637 Ok(route_pattern(reader, s, p, o)?.is_some())
2638 }
2639}
2640
2641struct RoutedPattern {
2644 dict: Dictionary,
2645 pattern: Pattern,
2646 permutation: IndexPermutation,
2647 header: Header,
2648 section: ByteRange,
2650}
2651
2652fn route_pattern<R: RangeReader>(
2655 reader: &R,
2656 s: Option<&str>,
2657 p: Option<&str>,
2658 o: Option<&str>,
2659) -> Result<Option<RoutedPattern>, FileError> {
2660 let head = reader.read_at(0, HEADER_LEN as u64)?;
2661 let header = Header::from_bytes(&head)?;
2662
2663 let dict_bytes = reader.read_at(header.dictionary_offset, header.dictionary_len)?;
2664 let dict = decode_dictionary_container(&dict_bytes, header.dict_codec)?;
2665
2666 let Some(pattern) = resolve_query_pattern(&dict, s, p, o) else {
2667 return Ok(None);
2668 };
2669 let permutation = GraphIndex::best_permutation(pattern);
2670 let section = locate_container_section_ranged(
2671 reader,
2672 header.root_dir_offset,
2673 header.root_dir_len,
2674 permutation.section_index(),
2675 NUM_PERMS as u64,
2676 )?;
2677 Ok(Some(RoutedPattern {
2678 dict,
2679 pattern,
2680 permutation,
2681 header,
2682 section,
2683 }))
2684}
2685
2686fn fetch_routed_matches<R: RangeReader>(
2691 reader: &R,
2692 routed: &RoutedPattern,
2693) -> Result<Vec<Triple>, FileError> {
2694 let dir = read_tile_directory_ranged(reader, routed.section)?;
2695 let [pa, _, _] = routed.permutation.order_pattern(routed.pattern);
2696 let codec = routed.header.block_codec;
2697 let mut out = Vec::new();
2698 match pa {
2699 Some(a) => {
2702 for e in dir.iter().filter(|e| e.min_a <= a && a <= e.max_a) {
2703 let bytes = reader.read_at(routed.section.offset + e.start, e.end - e.start)?;
2704 let tile = decompress(codec, &bytes)?;
2705 out.extend(GraphIndex::match_serialized_block(
2706 &tile,
2707 routed.permutation,
2708 routed.pattern,
2709 ));
2710 }
2711 }
2712 None => {
2715 if let (Some(first), Some(last)) = (dir.first(), dir.last()) {
2716 let base = first.start;
2717 let body = reader.read_at(routed.section.offset + base, last.end - base)?;
2718 for e in &dir {
2719 let tile = decompress(
2720 codec,
2721 &body[(e.start - base) as usize..(e.end - base) as usize],
2722 )?;
2723 out.extend(GraphIndex::match_serialized_block(
2724 &tile,
2725 routed.permutation,
2726 routed.pattern,
2727 ));
2728 }
2729 }
2730 }
2731 }
2732 out.sort_unstable();
2733 Ok(out)
2734}
2735
2736fn resolve_query_pattern(
2737 dict: &Dictionary,
2738 s: Option<&str>,
2739 p: Option<&str>,
2740 o: Option<&str>,
2741) -> Option<Pattern> {
2742 let sid = match s {
2743 Some(t) => Some(dict.subject_id(t)?),
2744 None => None,
2745 };
2746 let pid = match p {
2747 Some(t) => Some(dict.predicate_id(t)?),
2748 None => None,
2749 };
2750 let oid = match o {
2751 Some(t) => Some(dict.object_id(t)?),
2752 None => None,
2753 };
2754 Some((sid, pid, oid))
2755}
2756
2757#[must_use]
2762pub struct SummaryView {
2763 pub round: u32,
2764 pub summary: Vec<SuperEdge>,
2765 pub class_hierarchy: Vec<ClassNode>,
2767 pub level_rollups: Vec<LevelRollup>,
2769 pub level_links: Vec<LevelLinks>,
2771 pub descriptors: Vec<CommunityDescriptor>,
2773 pub subclass_cycles: Vec<Vec<String>>,
2775 pub disjoint_pairs: Vec<(String, String)>,
2777 pub equivalent_pairs: Vec<(String, String)>,
2779 dict: Dictionary,
2780}
2781
2782impl SummaryView {
2783 pub fn open_ranged<R: RangeReader>(reader: &R) -> Result<Option<Self>, FileError> {
2785 let head = reader.read_at(0, HEADER_LEN as u64)?;
2786 let header = Header::from_bytes(&head)?;
2787 if header.pyramid_meta_len == 0 {
2788 return Ok(None);
2789 }
2790
2791 let dict_bytes = reader.read_at(header.dictionary_offset, header.dictionary_len)?;
2792 let dict = decode_dictionary_container(&dict_bytes, header.dict_codec)?;
2793
2794 let mb = reader.read_at(header.pyramid_meta_offset, header.pyramid_meta_len)?;
2795 let meta =
2796 PyramidMeta::decode(&mb).map_err(|_| FileError::Container("malformed pyramid meta"))?;
2797
2798 Ok(Some(SummaryView {
2799 round: meta.round,
2800 summary: meta.summary,
2801 class_hierarchy: meta.class_hierarchy,
2802 level_rollups: meta.level_rollups,
2803 level_links: meta.level_links,
2804 descriptors: meta.descriptors,
2805 subclass_cycles: meta.subclass_cycles,
2806 disjoint_pairs: meta.disjoint_pairs,
2807 equivalent_pairs: meta.equivalent_pairs,
2808 dict,
2809 }))
2810 }
2811
2812 pub fn level_count(&self) -> usize {
2814 self.level_rollups.len()
2815 }
2816
2817 pub fn level_rollup(&self, k: usize) -> Option<&LevelRollup> {
2820 self.level_rollups.get(k)
2821 }
2822
2823 pub fn predicate_term(&self, id: u32) -> Option<String> {
2825 self.dict.predicate_term(id)
2826 }
2827
2828 pub fn predicate_total(&self, predicate: &str) -> u32 {
2831 match self.dict.predicate_id(predicate) {
2832 Some(pid) => self
2833 .summary
2834 .iter()
2835 .filter(|e| e.predicate == pid)
2836 .map(|e| e.count)
2837 .sum(),
2838 None => 0,
2839 }
2840 }
2841
2842 pub fn predicate_totals(&self) -> Vec<(String, u32)> {
2844 let mut by_pred: std::collections::BTreeMap<u32, u32> = std::collections::BTreeMap::new();
2845 for e in &self.summary {
2846 *by_pred.entry(e.predicate).or_default() += e.count;
2847 }
2848 let mut out: Vec<(String, u32)> = by_pred
2849 .into_iter()
2850 .filter_map(|(pid, c)| self.dict.predicate_term(pid).map(|t| (t, c)))
2851 .collect();
2852 out.sort_by(|a, b| b.1.cmp(&a.1).then_with(|| a.0.cmp(&b.0)));
2853 out
2854 }
2855
2856 pub fn community_count(&self) -> usize {
2858 let mut comms = std::collections::BTreeSet::new();
2859 for e in &self.summary {
2860 comms.insert(e.s_comm);
2861 comms.insert(e.o_comm);
2862 }
2863 comms.len()
2864 }
2865
2866 pub fn tbox_coherence(&self) -> Vec<crate::reason::Inconsistency> {
2881 schema_coherence(
2882 &self.class_hierarchy,
2883 &self.subclass_cycles,
2884 &self.disjoint_pairs,
2885 &self.equivalent_pairs,
2886 )
2887 }
2888
2889 pub fn tbox_is_coherent(&self) -> bool {
2892 self.tbox_coherence().is_empty()
2893 }
2894}
2895
2896pub fn schema_coherence(
2902 class_hierarchy: &[ClassNode],
2903 subclass_cycles: &[Vec<String>],
2904 disjoint_pairs: &[(String, String)],
2905 equivalent_pairs: &[(String, String)],
2906) -> Vec<crate::reason::Inconsistency> {
2907 use crate::reason::Inconsistency;
2908 use std::collections::{BTreeMap, BTreeSet, VecDeque};
2909 const MAX_REACH: usize = 100_000;
2910
2911 let mut out: Vec<Inconsistency> = Vec::new();
2912
2913 for cyc in subclass_cycles {
2914 let detail = if cyc.len() == 1 {
2915 format!("{} is rdfs:subClassOf itself (a cycle)", cyc[0])
2916 } else {
2917 format!(
2918 "classes {{{}}} are mutually rdfs:subClassOf (a cycle)",
2919 cyc.join(", ")
2920 )
2921 };
2922 out.push(Inconsistency {
2923 kind: "subclass-cycle",
2924 detail,
2925 });
2926 }
2927
2928 if !disjoint_pairs.is_empty() {
2929 let mut adj: BTreeMap<&str, Vec<&str>> = BTreeMap::new();
2931 for n in class_hierarchy {
2932 let e = adj.entry(n.class.as_str()).or_default();
2933 for p in &n.parents {
2934 e.push(p.as_str());
2935 }
2936 }
2937 for (a, b) in equivalent_pairs {
2938 adj.entry(a.as_str()).or_default().push(b.as_str());
2939 adj.entry(b.as_str()).or_default().push(a.as_str());
2940 }
2941
2942 let mut focuses: BTreeSet<&str> =
2944 class_hierarchy.iter().map(|n| n.class.as_str()).collect();
2945 for (a, b) in disjoint_pairs.iter().chain(equivalent_pairs) {
2946 focuses.insert(a.as_str());
2947 focuses.insert(b.as_str());
2948 }
2949
2950 let mut seen: BTreeSet<&str> = BTreeSet::new();
2951 for &c in &focuses {
2952 let mut reach: BTreeSet<&str> = BTreeSet::new();
2954 let mut q: VecDeque<&str> = VecDeque::new();
2955 reach.insert(c);
2956 q.push_back(c);
2957 while let Some(x) = q.pop_front() {
2958 if reach.len() > MAX_REACH {
2959 break;
2960 }
2961 if let Some(ns) = adj.get(x) {
2962 for &p in ns {
2963 if reach.insert(p) {
2964 q.push_back(p);
2965 }
2966 }
2967 }
2968 }
2969 for (x, y) in disjoint_pairs {
2970 if reach.contains(x.as_str()) && reach.contains(y.as_str()) && seen.insert(c) {
2971 out.push(Inconsistency {
2972 kind: "unsatisfiable-class",
2973 detail: format!(
2974 "{c} is a subclass of both {x} and {y}, which are \
2975 owl:disjointWith — no individual can be a {c}"
2976 ),
2977 });
2978 break;
2979 }
2980 }
2981 }
2982 }
2983
2984 out.sort_by(|a, b| (a.kind, &a.detail).cmp(&(b.kind, &b.detail)));
2985 out
2986}
2987
2988pub fn read_schema_coherence_ranged<R: RangeReader>(
2996 reader: &R,
2997) -> Result<Option<Vec<crate::reason::Inconsistency>>, FileError> {
2998 let head = reader.read_at(0, HEADER_LEN as u64)?;
2999 let header = Header::from_bytes(&head)?;
3000 if header.pyramid_meta_len == 0 {
3001 return Ok(None);
3002 }
3003 if header.schema_meta_len > 0 && (header.schema_meta_len as u64) <= header.pyramid_meta_len {
3007 let off =
3008 header.pyramid_meta_offset + header.pyramid_meta_len - header.schema_meta_len as u64;
3009 let block = reader.read_at(off, header.schema_meta_len as u64)?;
3010 let (hierarchy, cycles, disjoint, equivalent) = crate::meta::decode_schema_block(&block)
3011 .map_err(|_| FileError::Container("malformed schema block"))?;
3012 return Ok(Some(schema_coherence(
3013 &hierarchy,
3014 &cycles,
3015 &disjoint,
3016 &equivalent,
3017 )));
3018 }
3019 let mb = reader.read_at(header.pyramid_meta_offset, header.pyramid_meta_len)?;
3021 let meta =
3022 PyramidMeta::decode(&mb).map_err(|_| FileError::Container("malformed pyramid meta"))?;
3023 Ok(Some(schema_coherence(
3024 &meta.class_hierarchy,
3025 &meta.subclass_cycles,
3026 &meta.disjoint_pairs,
3027 &meta.equivalent_pairs,
3028 )))
3029}
3030
3031#[allow(clippy::type_complexity)]
3039pub fn read_schema_summary_ranged<R: RangeReader>(
3040 reader: &R,
3041) -> Result<Option<(Vec<(String, u64)>, Vec<(String, String, String, u64)>)>, FileError> {
3042 let head = reader.read_at(0, HEADER_LEN as u64)?;
3043 let header = Header::from_bytes(&head)?;
3044 if header.pyramid_meta_len == 0 {
3045 return Ok(None);
3046 }
3047 if header.schema_meta_len > 0 && (header.schema_meta_len as u64) <= header.pyramid_meta_len {
3048 let off =
3049 header.pyramid_meta_offset + header.pyramid_meta_len - header.schema_meta_len as u64;
3050 let block = reader.read_at(off, header.schema_meta_len as u64)?;
3051 let summary = crate::meta::decode_schema_block_summary(&block)
3052 .map_err(|_| FileError::Container("malformed schema block"))?;
3053 return Ok(Some(summary));
3054 }
3055 let mb = reader.read_at(header.pyramid_meta_offset, header.pyramid_meta_len)?;
3057 let meta =
3058 PyramidMeta::decode(&mb).map_err(|_| FileError::Container("malformed pyramid meta"))?;
3059 if meta.level_rollups.is_empty() && meta.level_links.is_empty() {
3060 return Ok(None);
3061 }
3062 let classes = meta
3063 .level_rollups
3064 .iter()
3065 .max_by_key(|r| r.depth)
3066 .map(|r| r.classes.clone())
3067 .unwrap_or_default();
3068 let relations = meta
3069 .level_links
3070 .iter()
3071 .max_by_key(|l| l.depth)
3072 .map(|l| {
3073 l.links
3074 .iter()
3075 .map(|c| {
3076 (
3077 c.s_class.clone(),
3078 c.predicate.clone(),
3079 c.o_class.clone(),
3080 c.count,
3081 )
3082 })
3083 .collect()
3084 })
3085 .unwrap_or_default();
3086 Ok(Some((classes, relations)))
3087}
3088
3089#[cfg(test)]
3090mod tests {
3091 use super::*;
3092 use crate::dictionary::DictionaryBuilder;
3093 use crate::index::GraphIndexBuilder;
3094
3095 #[test]
3096 fn read_coalesced_merges_within_gap_and_splits_beyond() {
3097 use crate::reader::{CountingReader, SliceReader};
3098 let bytes = vec![0u8; 4096];
3099 let ranges = [
3101 ByteRange { offset: 0, len: 16 },
3102 ByteRange {
3103 offset: 48,
3104 len: 16,
3105 },
3106 ByteRange {
3107 offset: 1088,
3108 len: 16,
3109 },
3110 ];
3111 let r = CountingReader::new(SliceReader::new(&bytes));
3113 let out = read_coalesced(&r, &ranges, 16).unwrap();
3114 assert_eq!(out.len(), 3);
3115 assert_eq!(r.requests(), 3);
3116 let r = CountingReader::new(SliceReader::new(&bytes));
3118 read_coalesced(&r, &ranges, 64).unwrap();
3119 assert_eq!(r.requests(), 2);
3120 let r = CountingReader::new(SliceReader::new(&bytes));
3122 read_coalesced(&r, &ranges, 4096).unwrap();
3123 assert_eq!(r.requests(), 1);
3124 }
3125
3126 fn build_image() -> Vec<u8> {
3127 let triples = [
3128 ("Alice", "knows", "Bob"),
3129 ("Bob", "knows", "Carol"),
3130 ("Alice", "age", "30"),
3131 ];
3132 let mut db = DictionaryBuilder::new();
3133 for (s, p, o) in triples {
3134 db.observe(s, p, o);
3135 }
3136 let dict = db.build();
3137
3138 let mut ib = GraphIndexBuilder::new();
3139 for (s, p, o) in triples {
3140 ib.push(dict.encode(s, p, o).unwrap());
3141 }
3142 let index = ib.build();
3143
3144 let (meta, levels) = build_pyramid_meta(&dict, &triples_ids(&dict), DEFAULT_TILE_BUDGET);
3145 write_file(&dict, &index, false, &meta, levels)
3146 }
3147
3148 fn triples_ids(dict: &Dictionary) -> Vec<(u32, u32, u32)> {
3149 [
3150 ("Alice", "knows", "Bob"),
3151 ("Bob", "knows", "Carol"),
3152 ("Alice", "age", "30"),
3153 ]
3154 .iter()
3155 .map(|(s, p, o)| dict.encode(s, p, o).unwrap())
3156 .collect()
3157 }
3158
3159 #[test]
3160 fn file_round_trips_header_and_counts() {
3161 let bytes = build_image();
3162 let rete = Rete::open(&bytes).unwrap();
3163 assert_eq!(rete.header().quad_count, 3);
3164 assert!(rete.header().term_count >= 5);
3165 let expected_codec = writer_codec();
3166 assert_eq!(rete.header().dict_codec, expected_codec);
3167 assert_eq!(rete.header().block_codec, expected_codec);
3168 assert_eq!(&bytes[bytes.len() - 4..], &MAGIC); }
3170
3171 #[test]
3176 fn multi_tile_file_round_trips_and_routes() {
3177 let triples: Vec<(String, String, String)> = (0..200)
3178 .map(|i| {
3179 (
3180 format!("<http://ex/s/{i}>"),
3181 format!("<http://ex/p/{}>", i % 5),
3182 format!("<http://ex/o/{}>", i % 23),
3183 )
3184 })
3185 .collect();
3186 let mut db = DictionaryBuilder::new();
3187 for (s, p, o) in &triples {
3188 db.observe(s, p, o);
3189 }
3190 let dict = db.build();
3191 let mut ib = GraphIndexBuilder::new().with_tile_budget(64);
3192 for (s, p, o) in &triples {
3193 ib.push(dict.encode(s, p, o).unwrap());
3194 }
3195 let index = ib.build();
3196 assert!(
3197 index.tile_sections()[0].len() > 3,
3198 "tiny budget must force many tiles"
3199 );
3200 let bytes = write_file(&dict, &index, false, &[], 0);
3201
3202 let rete = Rete::open(&bytes).unwrap();
3203 assert_eq!(rete.header().version, crate::header::CURRENT_FORMAT_VERSION);
3204 assert_eq!(rete.query(None, None, None).len(), 200);
3205 assert_eq!(rete.query(Some("<http://ex/s/7>"), None, None).len(), 1);
3206 assert_eq!(
3207 rete.query(None, Some("<http://ex/p/3>"), None).len(),
3208 40,
3209 "predicate extent spans tiles"
3210 );
3211 assert_eq!(
3212 rete.query(None, None, Some("<http://ex/o/22>")).len(),
3213 8 );
3215
3216 use crate::reader::SliceReader;
3218 let reader = SliceReader::new(&bytes);
3219 let routed = Rete::query_ranged(&reader, Some("<http://ex/s/7>"), None, None).unwrap();
3220 assert_eq!(routed.len(), 1);
3221 let routed = Rete::query_ranged(&reader, None, Some("<http://ex/p/3>"), None).unwrap();
3222 assert_eq!(routed.len(), 40);
3223 let routed = Rete::query_ranged(&reader, None, None, Some("<http://ex/o/22>")).unwrap();
3224 assert_eq!(routed.len(), 8);
3225 }
3226
3227 #[test]
3236 fn tile_directory_offsets_survive_past_4gib() {
3237 let mut dir = Vec::new();
3238 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);
3244 write_uvarint(&mut dir, 2 << 30); let total = dir.len() as u64 + (3u64 << 30) + (2u64 << 30) + 64;
3246 let entries = parse_tile_directory(&dir, total).unwrap();
3247 assert_eq!(entries.len(), 2);
3248 assert_eq!(entries[1].start, dir.len() as u64 + (3u64 << 30));
3249 assert!(
3250 entries[1].end > u32::MAX as u64,
3251 "tail tile sits past 4 GiB"
3252 );
3253 assert!(parse_tile_directory(&dir, 1 << 20).is_err());
3255 }
3256
3257 #[test]
3258 fn tile_synopsis_trailer_round_trips() {
3259 let mut ib = GraphIndexBuilder::new().with_tile_budget(64);
3260 for i in 0..200u32 {
3261 ib.push((i, i % 7, i % 13));
3262 }
3263 let index = ib.build();
3264 let tiles = index.tile_sections()[0];
3265 assert!(tiles.len() > 3, "tiny budget forces many tiles");
3266
3267 let payload = encode_tiled_section(tiles, CODEC_NONE);
3268 let dir = parse_tile_directory(&payload, payload.len() as u64).unwrap();
3269 assert_eq!(dir.len(), tiles.len());
3270 let trailer_start = dir.iter().map(|e| e.end).max().unwrap();
3272 assert!(
3273 trailer_start < payload.len() as u64,
3274 "a trailer follows the tiles"
3275 );
3276 for e in &dir {
3277 assert!(
3278 e.end <= payload.len() as u64,
3279 "tiles still located within the payload"
3280 );
3281 }
3282 let syn = parse_tile_synopsis(&payload, trailer_start as usize, dir.len()).unwrap();
3283 for (e, (min_b, max_b, min_c, max_c)) in dir.iter().zip(syn) {
3284 let block = decompress(CODEC_NONE, &payload[e.start as usize..e.end as usize]).unwrap();
3285 let z = *crate::triples::TripleBlock::parse(&block).unwrap().zone();
3286 assert_eq!(
3287 (min_b, max_b, min_c, max_c),
3288 (z.min_b, z.max_b, z.min_c, z.max_c),
3289 "synopsis equals the tile's own zone"
3290 );
3291 }
3292 }
3293
3294 #[test]
3298 fn tile_synopsis_lazy_matches_reference_every_shape() {
3299 use crate::reader::{CountingReader, SliceReader};
3300 let triples: Vec<(String, String, String)> = (0..200u32)
3301 .map(|i| {
3302 (
3303 format!("<http://ex/s/{i:04}>"),
3304 format!("<http://ex/p/{}>", i % 7),
3305 format!("<http://ex/o/{:04}>", i % 13),
3306 )
3307 })
3308 .collect();
3309 let mut db = DictionaryBuilder::new();
3310 for (s, p, o) in &triples {
3311 db.observe(s, p, o);
3312 }
3313 let dict = db.build();
3314 let mut ib = GraphIndexBuilder::new().with_tile_budget(64);
3315 for (s, p, o) in &triples {
3316 ib.push(dict.encode(s, p, o).unwrap());
3317 }
3318 let bytes = write_file(&dict, &ib.build(), false, &[], 0);
3319
3320 let eager = Rete::open(&bytes).unwrap();
3321 assert!(
3322 eager.header().has_tile_synopsis(),
3323 "new files set the synopsis flag"
3324 );
3325
3326 let leaked: &'static [u8] = Box::leak(bytes.clone().into_boxed_slice());
3329 let reader = std::sync::Arc::new(CountingReader::new(SliceReader::new(leaked)));
3330 let lazy = Rete::open_ranged_lazy(reader).unwrap();
3331
3332 let brute = |s: Option<&str>, p: Option<&str>, o: Option<&str>| {
3333 let mut v: Vec<(String, String, String)> = triples
3334 .iter()
3335 .filter(|(a, b, c)| {
3336 s.is_none_or(|x| x == a) && p.is_none_or(|x| x == b) && o.is_none_or(|x| x == c)
3337 })
3338 .cloned()
3339 .collect();
3340 v.sort();
3341 v
3342 };
3343 let sv = [
3345 None,
3346 Some("<http://ex/s/0007>"),
3347 Some("<http://ex/s/0130>"),
3348 Some("<http://ex/s/9999>"),
3349 ];
3350 let pv = [
3351 None,
3352 Some("<http://ex/p/3>"),
3353 Some("<http://ex/p/6>"),
3354 Some("<http://ex/p/999>"),
3355 ];
3356 let ov = [
3357 None,
3358 Some("<http://ex/o/0000>"),
3359 Some("<http://ex/o/0012>"),
3360 Some("<http://ex/o/9999>"),
3361 ];
3362 for &s in &sv {
3363 for &p in &pv {
3364 for &o in &ov {
3365 let mut e = eager.query(s, p, o);
3366 e.sort();
3367 let mut l = lazy.query(s, p, o);
3368 l.sort();
3369 let r = brute(s, p, o);
3370 assert_eq!(e, r, "eager {s:?} {p:?} {o:?}");
3371 assert_eq!(l, r, "lazy {s:?} {p:?} {o:?} — synopsis over-pruned");
3372 }
3373 }
3374 }
3375 assert!(!lazy.index_incomplete(), "no lazy fetch failed");
3376 }
3377
3378 #[cfg(test)]
3381 fn build_text_indexed(triples: &[(String, String, String)]) -> Vec<u8> {
3382 let mut db = DictionaryBuilder::new();
3383 for (s, p, o) in triples {
3384 db.observe(s, p, o);
3385 }
3386 let dict = db.build();
3387 let mut ib = GraphIndexBuilder::new().with_tile_budget(64);
3388 let mut id_triples: Vec<(u32, u32, u32)> = Vec::with_capacity(triples.len());
3389 for (s, p, o) in triples {
3390 let t = dict.encode(s, p, o).unwrap();
3391 ib.push(t);
3392 id_triples.push(t);
3393 }
3394 let index = ib.build();
3395 let text_index = compute_text_index(&dict, &id_triples);
3396 assert!(
3397 !text_index.is_empty(),
3398 "literals should produce a text index"
3399 );
3400 write_dataset_with_metadata(&dict, &index, &[], false, &[], 0, &[], &text_index)
3401 }
3402
3403 #[test]
3407 fn text_index_eager_matches_brute_force() {
3408 let triples: Vec<(String, String, String)> = vec![
3409 (
3410 "<http://ex/s0>",
3411 "<http://ex/label>",
3412 "\"alpha glucose phosphate\"",
3413 ),
3414 ("<http://ex/s1>", "<http://ex/label>", "\"beta Glucose\""),
3415 ("<http://ex/s2>", "<http://ex/label>", "\"gamma fructose\""),
3416 (
3417 "<http://ex/s3>",
3418 "<http://ex/note>",
3419 "\"einstein relativity\"",
3420 ),
3421 (
3422 "<http://ex/s4>",
3423 "<http://ex/ref>",
3424 "<http://ex/not-a-literal>",
3425 ),
3426 ]
3427 .into_iter()
3428 .map(|(s, p, o)| (s.to_string(), p.to_string(), o.to_string()))
3429 .collect();
3430 let bytes = build_text_indexed(&triples);
3431 let rete = Rete::open(&bytes).unwrap();
3432 assert!(rete.has_text_index());
3433
3434 let brute = |words: &[&str]| -> Vec<String> {
3436 let mut v: Vec<String> = triples
3437 .iter()
3438 .filter(|(_, _, o)| {
3439 crate::terms::is_literal(o)
3440 && words.iter().all(|w| {
3441 let wl = w.to_lowercase();
3442 crate::terms::literal_lexical(o)
3443 .unwrap()
3444 .split(|c: char| !c.is_alphanumeric())
3445 .any(|t| t.to_lowercase() == wl)
3446 })
3447 })
3448 .map(|(s, _, _)| s.clone())
3449 .collect();
3450 v.sort();
3451 v.dedup();
3452 v
3453 };
3454
3455 let mut got = rete.text_search(&["glucose"], None, 0);
3456 got.sort();
3457 assert_eq!(got, brute(&["glucose"]), "case-insensitive single word");
3458
3459 let mut got = rete.text_search(&["glucose", "phosphate"], None, 0);
3461 got.sort();
3462 assert_eq!(got, brute(&["glucose", "phosphate"]));
3463
3464 assert!(rete.text_search(&["zzznope"], None, 0).is_empty());
3466
3467 let got = rete.text_search(&[], Some("ein"), 0);
3469 assert_eq!(got, vec!["<http://ex/s3>".to_string()]);
3470
3471 let mut db = DictionaryBuilder::new();
3473 for (s, p, o) in &triples {
3474 db.observe(s, p, o);
3475 }
3476 let dict = db.build();
3477 let mut ib = GraphIndexBuilder::new();
3478 for (s, p, o) in &triples {
3479 ib.push(dict.encode(s, p, o).unwrap());
3480 }
3481 let plain = write_dataset(&dict, &ib.build(), &[], false, &[], 0);
3482 let plain_rete = Rete::open(&plain).unwrap();
3483 assert!(!plain_rete.has_text_index());
3484 assert!(plain_rete.text_search(&["glucose"], None, 0).is_empty());
3485 }
3486
3487 #[test]
3491 fn text_index_lazy_faults_only_queried_postings() {
3492 use crate::reader::{CountingReader, SliceReader};
3493 let mut triples: Vec<(String, String, String)> = (0..300u32)
3496 .map(|i| {
3497 (
3498 format!("<http://ex/s/{i:04}>"),
3499 "<http://ex/label>".to_string(),
3500 format!("\"common word number {i}\""),
3501 )
3502 })
3503 .collect();
3504 for i in [3u32, 77, 250] {
3505 triples.push((
3506 format!("<http://ex/s/{i:04}>"),
3507 "<http://ex/tag>".to_string(),
3508 "\"raretoken\"".to_string(),
3509 ));
3510 }
3511 let bytes = build_text_indexed(&triples);
3512 let eager = Rete::open(&bytes).unwrap();
3513 let mut want = eager.text_search(&["raretoken"], None, 0);
3514 want.sort();
3515 assert_eq!(want.len(), 3);
3516
3517 let leaked: &'static [u8] = Box::leak(bytes.clone().into_boxed_slice());
3518 let reader = std::sync::Arc::new(CountingReader::new(SliceReader::new(leaked)));
3519 let lazy = Rete::open_ranged_lazy(reader.clone()).unwrap();
3520 let before = reader.bytes_read();
3523 let mut got = lazy.text_search(&["raretoken"], None, 0);
3524 got.sort();
3525 assert_eq!(got, want, "lazy search matches eager");
3526 let pulled = reader.bytes_read() - before;
3527 let ti_len = eager.header().text_index_len;
3530 assert!(
3531 pulled < ti_len,
3532 "search pulled {pulled} B but the section is {ti_len} B — faulted too much"
3533 );
3534 assert!(!lazy.index_incomplete());
3535 }
3536
3537 #[test]
3542 fn text_index_is_tamper_evident_and_verifies() {
3543 let triples: Vec<(String, String, String)> = vec![(
3544 "<http://ex/s0>".to_string(),
3545 "<http://ex/label>".to_string(),
3546 "\"alpha glucose phosphate\"".to_string(),
3547 )];
3548 let bytes = build_text_indexed(&triples);
3549 let header = Rete::open(&bytes).unwrap().header().clone();
3550 assert!(header.text_index_len > 0);
3551 assert!(verify(&bytes).unwrap(), "a text-indexed build must verify");
3552
3553 let mut tampered = bytes.clone();
3554 tampered[header.text_index_offset as usize] ^= 0xff;
3555 assert!(
3556 !verify(&tampered).unwrap(),
3557 "tampering with the text index must break verify()"
3558 );
3559 }
3560
3561 #[test]
3565 fn synopsis_cuts_remote_fetch_bytes() {
3566 use crate::header::FLAG_TILE_SYNOPSIS;
3567 use crate::reader::{CountingReader, SliceReader};
3568
3569 let triples: Vec<(String, String, String)> = (0..400u32)
3573 .map(|i| {
3574 (
3575 format!("<http://ex/s/{i:04}>"),
3576 "<http://ex/p>".to_string(),
3577 format!("<http://ex/o/{i:04}>"),
3578 )
3579 })
3580 .collect();
3581 let mut db = DictionaryBuilder::new();
3582 for (s, p, o) in &triples {
3583 db.observe(s, p, o);
3584 }
3585 let dict = db.build();
3586 let mut ib = GraphIndexBuilder::new().with_tile_budget(64);
3587 for (s, p, o) in &triples {
3588 ib.push(dict.encode(s, p, o).unwrap());
3589 }
3590 let bytes = write_file(&dict, &ib.build(), false, &[], 0);
3591
3592 let q = (Some("<http://ex/s/0395>"), None, Some("<http://ex/o/0005>"));
3596 let query_bytes = |image: &[u8]| -> (u64, usize) {
3600 let leaked: &'static [u8] = Box::leak(image.to_vec().into_boxed_slice());
3601 let reader = std::sync::Arc::new(CountingReader::new(SliceReader::new(leaked)));
3602 let rete = Rete::open_ranged_lazy(reader.clone()).unwrap();
3603 let before = reader.bytes_read(); let n = rete.query(q.0, q.1, q.2).len();
3605 assert!(!rete.index_incomplete());
3606 (reader.bytes_read() - before, n)
3607 };
3608
3609 let (on_bytes, on_n) = query_bytes(&bytes);
3610 let mut off = bytes.clone();
3612 off[5] &= !FLAG_TILE_SYNOPSIS;
3613 let (off_bytes, off_n) = query_bytes(&off);
3614
3615 assert_eq!(on_n, 0, "the pair never co-occurs");
3616 assert_eq!(off_n, 0, "same answer without the synopsis");
3617 assert!(
3621 on_bytes < off_bytes,
3622 "synopsis skips the routed tile fetch: {on_bytes} < {off_bytes}"
3623 );
3624 }
3625
3626 #[test]
3632 fn double_bound_object_join_eager_matches_lazy() {
3633 use crate::reader::SliceReader;
3634 let occ = "<http://ex/occ>";
3635 let phys = "<http://ex/physicist>";
3636 let phil = "<http://ex/philosopher>";
3637 let label = "<http://www.w3.org/2000/01/rdf-schema#label>";
3638 let mut triples: Vec<(String, String, String)> = Vec::new();
3640 for i in 0..20u32 {
3641 triples.push((format!("<http://ex/p/{i:02}>"), occ.into(), phys.into()));
3642 if i < 10 {
3643 triples.push((format!("<http://ex/p/{i:02}>"), occ.into(), phil.into()));
3644 }
3645 triples.push((
3646 format!("<http://ex/p/{i:02}>"),
3647 label.into(),
3648 format!("\"Name {i:02}\""),
3649 ));
3650 }
3651 let mut db = DictionaryBuilder::new();
3652 for (s, p, o) in &triples {
3653 db.observe(s, p, o);
3654 }
3655 let dict = db.build();
3656 let mut ib = GraphIndexBuilder::new().with_tile_budget(16);
3657 for (s, p, o) in &triples {
3658 ib.push(dict.encode(s, p, o).unwrap());
3659 }
3660 let bytes = write_file(&dict, &ib.build(), false, &[], 0);
3661
3662 let q = "SELECT ?l WHERE { \
3663 ?p <http://ex/occ> <http://ex/physicist> ; \
3664 <http://ex/occ> <http://ex/philosopher> ; \
3665 <http://www.w3.org/2000/01/rdf-schema#label> ?l }";
3666 let run = |rete: &Rete| -> Vec<String> {
3667 let (_, sols) = crate::eval_sparql(rete, q).unwrap();
3668 let mut v: Vec<String> = sols.iter().filter_map(|b| b.get("l").cloned()).collect();
3669 v.sort();
3670 v
3671 };
3672
3673 let eager_rows = run(&Rete::open(&bytes).unwrap());
3674 let leaked: &'static [u8] = Box::leak(bytes.clone().into_boxed_slice());
3675 let lazy = Rete::open_ranged_lazy(std::sync::Arc::new(SliceReader::new(leaked))).unwrap();
3676 let lazy_rows = run(&lazy);
3677 assert!(!lazy.index_incomplete());
3678
3679 assert_eq!(eager_rows.len(), 10, "the 10 physicist∩philosopher labels");
3680 assert_eq!(eager_rows, lazy_rows, "eager and lazy must agree exactly");
3681 }
3682
3683 #[test]
3687 fn multi_chunk_dictionary_round_trips() {
3688 let mut db = DictionaryBuilder::new();
3689 let term = |i: u32| format!("<http://example.org/some/long/prefix/entity/{i:06}>");
3690 for i in 0..6000u32 {
3691 db.observe(&term(i), "<http://ex/p>", &term(i + 1));
3692 }
3693 let dict = db.build();
3694 let mut ib = GraphIndexBuilder::new();
3695 for i in 0..6000u32 {
3696 ib.push(
3697 dict.encode(&term(i), "<http://ex/p>", &term(i + 1))
3698 .unwrap(),
3699 );
3700 }
3701 let bytes = write_file(&dict, &ib.build(), false, &[], 0);
3702 let rete = Rete::open(&bytes).unwrap();
3703 let d = rete.dictionary();
3704 assert_eq!(d.term_count(), dict.term_count());
3705 for i in (0..6000).step_by(97).chain([0, 1, 5999, 6000]) {
3706 let t = term(i);
3707 let sid = dict.subject_id(&t);
3708 assert_eq!(d.subject_id(&t), sid, "subject_id({t})");
3709 if let Some(id) = sid {
3710 assert_eq!(d.subject_term(id).as_deref(), Some(t.as_str()));
3711 }
3712 let oid = dict.object_id(&t);
3713 assert_eq!(d.object_id(&t), oid, "object_id({t})");
3714 }
3715 assert_eq!(d.subject_id("<http://example.org/absent>"), None);
3716 assert_eq!(d.predicate_id("<http://ex/p>"), Some(1));
3717 assert_eq!(d.predicate_term(1).as_deref(), Some("<http://ex/p>"));
3718 }
3719
3720 #[test]
3721 #[cfg(feature = "compression")]
3722 fn compression_shrinks_repetitive_data() {
3723 let mut db = DictionaryBuilder::new();
3727 let triples: Vec<(String, String, String)> = (0..500)
3728 .map(|i| {
3729 (
3730 format!("<http://example.org/entity/{i}>"),
3731 "<http://example.org/p/relatedTo>".to_string(),
3732 format!("<http://example.org/entity/{}>", (i + 1) % 500),
3733 )
3734 })
3735 .collect();
3736 for (s, p, o) in &triples {
3737 db.observe(s, p, o);
3738 }
3739 let dict = db.build();
3740 let mut ib = GraphIndexBuilder::new();
3741 for (s, p, o) in &triples {
3742 ib.push(dict.encode(s, p, o).unwrap());
3743 }
3744 let bytes = write_file(&dict, &ib.build(), false, &[], 0);
3745
3746 let raw: usize = triples
3747 .iter()
3748 .map(|(s, p, o)| s.len() + p.len() + o.len())
3749 .sum();
3750 assert!(
3751 bytes.len() < raw / 2,
3752 "expected strong compression: file {} vs raw terms {raw}",
3753 bytes.len()
3754 );
3755
3756 let rete = Rete::open(&bytes).unwrap();
3758 let r = rete.query(Some("<http://example.org/entity/0>"), None, None);
3759 assert_eq!(r.len(), 1);
3760 assert_eq!(r[0].2, "<http://example.org/entity/1>");
3761 }
3762
3763 fn big_file_with_pyramid() -> Vec<u8> {
3764 let triples: Vec<(String, String, String)> = (0..300)
3766 .map(|i| {
3767 (
3768 format!("<http://ex/e{i}>"),
3769 "<http://ex/next>".to_string(),
3770 format!("<http://ex/e{}>", (i + 1) % 300),
3771 )
3772 })
3773 .collect();
3774 let mut db = DictionaryBuilder::new();
3775 for (s, p, o) in &triples {
3776 db.observe(s, p, o);
3777 }
3778 let dict = db.build();
3779 let ids: Vec<_> = triples
3780 .iter()
3781 .map(|(s, p, o)| dict.encode(s, p, o).unwrap())
3782 .collect();
3783 let mut ib = GraphIndexBuilder::new();
3784 for &t in &ids {
3785 ib.push(t);
3786 }
3787 let (meta, levels) = build_pyramid_meta(&dict, &ids, DEFAULT_TILE_BUDGET);
3788 write_file(&dict, &ib.build(), false, &meta, levels)
3789 }
3790
3791 #[test]
3792 fn ranged_open_is_minimal_and_correct() {
3793 use crate::reader::{CountingReader, SliceReader};
3794 let bytes = big_file_with_pyramid();
3795
3796 let full = CountingReader::new(SliceReader::new(&bytes));
3797 let rete = Rete::open_ranged(&full).unwrap();
3798 assert!(full.requests() <= 4, "requests = {}", full.requests());
3800 assert_eq!(
3801 rete.query(Some("<http://ex/e0>"), None, None)[0].2,
3802 "<http://ex/e1>"
3803 );
3804
3805 let summ_reader = CountingReader::new(SliceReader::new(&bytes));
3807 let view = SummaryView::open_ranged(&summ_reader).unwrap().unwrap();
3808 assert!(!view.summary.is_empty());
3809 assert!(
3810 summ_reader.bytes_read() < bytes.len() as u64,
3811 "summary read {} of {} bytes",
3812 summ_reader.bytes_read(),
3813 bytes.len()
3814 );
3815 assert!(summ_reader.bytes_read() < full.bytes_read());
3817 }
3818
3819 #[test]
3820 fn content_hash_is_set_and_verifies() {
3821 let bytes = build_image();
3822 let rete = Rete::open(&bytes).unwrap();
3823 assert_ne!(
3824 rete.header().content_hash,
3825 [0u8; 16],
3826 "hash must be populated"
3827 );
3828 assert!(verify(&bytes).unwrap(), "freshly built file verifies");
3829
3830 assert_eq!(
3832 Rete::open(&build_image()).unwrap().header().content_hash,
3833 rete.header().content_hash
3834 );
3835
3836 let mut tampered = bytes.clone();
3838 let last = tampered.len() - 5; tampered[last] ^= 0xff;
3840 assert!(!verify(&tampered).unwrap());
3841 }
3842
3843 fn build_with_metadata(meta: &[u8]) -> Vec<u8> {
3845 let triples = [
3846 ("Alice", "knows", "Bob"),
3847 ("Bob", "knows", "Carol"),
3848 ("Alice", "age", "30"),
3849 ];
3850 let mut db = DictionaryBuilder::new();
3851 for (s, p, o) in triples {
3852 db.observe(s, p, o);
3853 }
3854 let dict = db.build();
3855 let ids: Vec<_> = triples
3856 .iter()
3857 .map(|(s, p, o)| dict.encode(s, p, o).unwrap())
3858 .collect();
3859 let mut ib = GraphIndexBuilder::new();
3860 for &t in &ids {
3861 ib.push(t);
3862 }
3863 let (pmeta, levels) = build_pyramid_meta(&dict, &ids, DEFAULT_TILE_BUDGET);
3864 write_dataset_with_metadata(&dict, &ib.build(), &[], false, &pmeta, levels, meta, &[])
3865 }
3866
3867 #[test]
3868 fn metadata_round_trips_and_shifts_offsets() {
3869 let card = br#"{"title":"My Dataset"}"#;
3870 let bytes = build_with_metadata(card);
3871 let rete = Rete::open(&bytes).unwrap();
3872
3873 assert_eq!(rete.metadata(), Some(card.as_slice()));
3875 let h = rete.header();
3876 assert_eq!(h.metadata_offset, HEADER_LEN as u64);
3877 assert_eq!(h.metadata_len, card.len() as u64);
3878 assert_eq!(h.dictionary_offset, HEADER_LEN as u64 + card.len() as u64);
3880
3881 assert_eq!(
3883 rete.query(Some("Bob"), Some("knows"), Some("Carol")).len(),
3884 1
3885 );
3886 assert!(verify(&bytes).unwrap());
3888 }
3889
3890 #[test]
3891 fn empty_metadata_is_byte_identical_to_plain_writer() {
3892 assert_eq!(
3895 build_with_metadata(&[]),
3896 build_image(),
3897 "empty-metadata output must equal the plain writer byte-for-byte"
3898 );
3899 }
3900
3901 #[test]
3902 fn metadata_is_tamper_evident() {
3903 let card = br#"{"title":"x"}"#;
3904 let mut bytes = build_with_metadata(card);
3905 assert!(verify(&bytes).unwrap());
3906 bytes[HEADER_LEN + 2] ^= 0xff;
3908 assert!(
3909 !verify(&bytes).unwrap(),
3910 "tampering with the card must break verify()"
3911 );
3912 }
3913
3914 #[test]
3915 fn ranged_opens_do_not_fetch_metadata() {
3916 use crate::reader::{CountingReader, SliceReader};
3917 let card = vec![0xABu8; 512]; let bytes = build_with_metadata(&card);
3919 let total = bytes.len() as u64;
3920
3921 let r = CountingReader::new(SliceReader::new(&bytes));
3923 let rete = Rete::open_ranged(&r).unwrap();
3924 assert!(
3925 rete.metadata().is_none(),
3926 "open_ranged must not load the card"
3927 );
3928 assert!(r.requests() <= 4, "requests = {}", r.requests());
3929 assert!(
3930 r.bytes_read() <= total - card.len() as u64,
3931 "read {} of {} bytes; the {}-byte card must be skipped",
3932 r.bytes_read(),
3933 total,
3934 card.len()
3935 );
3936
3937 let rs = CountingReader::new(SliceReader::new(&bytes));
3939 let view = SummaryView::open_ranged(&rs).unwrap().unwrap();
3940 assert!(!view.summary.is_empty());
3941 assert!(rs.bytes_read() <= total - card.len() as u64);
3942 }
3943
3944 #[test]
3945 fn metadata_ranged_fetches_only_header_and_card() {
3946 use crate::reader::{CountingReader, SliceReader};
3947 let card = vec![0xCDu8; 384];
3950 let bytes = build_with_metadata(&card);
3951
3952 let r = CountingReader::new(SliceReader::new(&bytes));
3953 let got = read_metadata_ranged(&r).unwrap().unwrap();
3954 assert_eq!(got, card, "the card reads back verbatim");
3955 assert_eq!(r.requests(), 2, "exactly header + metadata ranges");
3956 assert_eq!(
3957 r.bytes_read(),
3958 HEADER_LEN as u64 + card.len() as u64,
3959 "no dictionary/index/pyramid bytes are touched"
3960 );
3961
3962 let plain = build_image();
3964 let rp = CountingReader::new(SliceReader::new(&plain));
3965 assert!(read_metadata_ranged(&rp).unwrap().is_none());
3966 assert_eq!(rp.requests(), 1, "header only for a cardless file");
3967 assert_eq!(rp.bytes_read(), HEADER_LEN as u64);
3968 }
3969
3970 #[test]
3971 fn schema_summary_groups_by_type() {
3972 let rt = RDF_TYPE;
3973 let bytes = build_from(&[
3974 ("Alice", rt, "Person"),
3975 ("Bob", rt, "Person"),
3976 ("NYC", rt, "City"),
3977 ("Alice", "knows", "Bob"),
3978 ("Alice", "livesIn", "NYC"),
3979 ("Alice", "name", "\"Alice\""),
3980 ]);
3981 let rete = Rete::open(&bytes).unwrap();
3982 let summary = schema_summary(&rete);
3983 assert!(summary.contains(&("Person".into(), "knows".into(), "Person".into(), 1)));
3985 assert!(summary.contains(&("Person".into(), "livesIn".into(), "City".into(), 1)));
3986 assert!(summary.contains(&("Person".into(), "name".into(), "(literal)".into(), 1)));
3987 assert!(!summary.iter().any(|(_, p, _, _)| p == RDF_TYPE));
3989
3990 let classes = schema_classes(&rete);
3992 assert_eq!(
3993 classes,
3994 vec![("Person".into(), 2u32), ("City".into(), 1u32)]
3995 );
3996 }
3997
3998 fn build_from(triples: &[(&str, &str, &str)]) -> Vec<u8> {
3999 let mut db = DictionaryBuilder::new();
4000 for (s, p, o) in triples {
4001 db.observe(s, p, o);
4002 }
4003 let dict = db.build();
4004 let mut ib = GraphIndexBuilder::new();
4005 for (s, p, o) in triples {
4006 ib.push(dict.encode(s, p, o).unwrap());
4007 }
4008 write_file(&dict, &ib.build(), false, &[], 0)
4009 }
4010
4011 fn build_with_pyramid(triples: &[(&str, &str, &str)]) -> Vec<u8> {
4013 let mut db = DictionaryBuilder::new();
4014 for (s, p, o) in triples {
4015 db.observe(s, p, o);
4016 }
4017 let dict = db.build();
4018 let encoded: Vec<_> = triples
4019 .iter()
4020 .map(|(s, p, o)| dict.encode(s, p, o).unwrap())
4021 .collect();
4022 let mut ib = GraphIndexBuilder::new();
4023 for t in &encoded {
4024 ib.push(*t);
4025 }
4026 let (meta, levels) = build_pyramid_meta(&dict, &encoded, DEFAULT_TILE_BUDGET);
4027 write_dataset(&dict, &ib.build(), &[], false, &meta, levels)
4028 }
4029
4030 #[test]
4031 fn tbox_coherence_flags_unsatisfiable_class_index_free() {
4032 use crate::reader::{CountingReader, SliceReader};
4033 let rt = RDF_TYPE;
4034 let sub = "<http://www.w3.org/2000/01/rdf-schema#subClassOf>";
4035 let disj = "<http://www.w3.org/2002/07/owl#disjointWith>";
4036 let bytes = build_with_pyramid(&[
4039 ("<http://ex/C>", sub, "<http://ex/D>"),
4040 ("<http://ex/C>", sub, "<http://ex/E>"),
4041 ("<http://ex/D>", disj, "<http://ex/E>"),
4042 ("<http://ex/x>", rt, "<http://ex/C>"),
4043 ]);
4044
4045 let r = CountingReader::new(SliceReader::new(&bytes));
4046 let view = SummaryView::open_ranged(&r).unwrap().unwrap();
4047 let points = view.tbox_coherence();
4048 assert!(
4049 points
4050 .iter()
4051 .any(|i| i.kind == "unsatisfiable-class" && i.detail.contains("http://ex/C>")),
4052 "expected C unsatisfiable from the schema alone, got {points:?}"
4053 );
4054
4055 let header = Header::from_bytes(&bytes[..HEADER_LEN]).unwrap();
4058 assert!(
4059 r.bytes_read() <= bytes.len() as u64 - header.root_dir_len,
4060 "tbox_coherence must not read the triple index"
4061 );
4062 }
4063
4064 #[test]
4065 fn schema_coherence_reads_only_the_schema_block() {
4066 use crate::reader::{CountingReader, SliceReader};
4067 let rt = RDF_TYPE;
4068 let sub = "<http://www.w3.org/2000/01/rdf-schema#subClassOf>";
4069 let disj = "<http://www.w3.org/2002/07/owl#disjointWith>";
4070 let mut triples: Vec<(String, String, String)> = vec![
4074 ("<http://ex/C>".into(), sub.into(), "<http://ex/D>".into()),
4075 ("<http://ex/C>".into(), sub.into(), "<http://ex/E>".into()),
4076 ("<http://ex/D>".into(), disj.into(), "<http://ex/E>".into()),
4077 ];
4078 for i in 0..500 {
4079 let s = format!("<http://ex/x{i}>");
4080 triples.push((s.clone(), rt.into(), "<http://ex/C>".into()));
4081 triples.push((
4082 s,
4083 "<http://ex/label>".into(),
4084 format!("\"unique label {i}\""),
4085 ));
4086 }
4087 let trefs: Vec<(&str, &str, &str)> = triples
4088 .iter()
4089 .map(|(s, p, o)| (s.as_str(), p.as_str(), o.as_str()))
4090 .collect();
4091 let bytes = build_with_pyramid(&trefs);
4092
4093 let header = Header::from_bytes(&bytes[..HEADER_LEN]).unwrap();
4094 assert!(
4095 header.schema_meta_len > 0,
4096 "the writer recorded a schema-block length"
4097 );
4098 assert!(
4099 (header.schema_meta_len as u64) < header.pyramid_meta_len,
4100 "schema block ({}) should be far smaller than the whole pyramid-meta ({})",
4101 header.schema_meta_len,
4102 header.pyramid_meta_len
4103 );
4104
4105 let r = CountingReader::new(SliceReader::new(&bytes));
4106 let points = read_schema_coherence_ranged(&r).unwrap().unwrap();
4107 assert!(points.iter().any(|i| i.kind == "unsatisfiable-class"));
4108 assert!(
4110 r.bytes_read() <= HEADER_LEN as u64 + header.schema_meta_len as u64,
4111 "read {} bytes; expected <= header + schema block ({})",
4112 r.bytes_read(),
4113 HEADER_LEN as u64 + header.schema_meta_len as u64
4114 );
4115 }
4116
4117 #[test]
4118 fn tbox_coherence_clean_schema_is_coherent() {
4119 use crate::reader::SliceReader;
4120 let rt = RDF_TYPE;
4121 let sub = "<http://www.w3.org/2000/01/rdf-schema#subClassOf>";
4122 let bytes = build_with_pyramid(&[
4123 ("<http://ex/Dog>", sub, "<http://ex/Animal>"),
4124 ("<http://ex/x>", rt, "<http://ex/Dog>"),
4125 ]);
4126 let view = SummaryView::open_ranged(&SliceReader::new(&bytes))
4127 .unwrap()
4128 .unwrap();
4129 assert!(view.tbox_is_coherent(), "a plain hierarchy is coherent");
4130 }
4131
4132 #[test]
4133 fn named_graphs_round_trip() {
4134 let all = [
4136 ("Alice", "knows", "Bob"), ("Bob", "age", "30"), ];
4139 let mut db = DictionaryBuilder::new();
4140 for (s, p, o) in all {
4141 db.observe(s, p, o);
4142 }
4143 let dict = db.build();
4144
4145 let mut def = GraphIndexBuilder::new();
4146 def.push(dict.encode("Alice", "knows", "Bob").unwrap());
4147 let mut g1 = GraphIndexBuilder::new();
4148 g1.push(dict.encode("Bob", "age", "30").unwrap());
4149
4150 let named = vec![("http://ex/g1".to_string(), g1.build())];
4151 let bytes = write_dataset(&dict, &def.build(), &named, true, &[], 0);
4152
4153 assert!(verify(&bytes).unwrap());
4154 let rete = Rete::open(&bytes).unwrap();
4155 assert_eq!(rete.graph_names(), vec!["http://ex/g1"]);
4156 let gi = rete.graph_index("http://ex/g1").unwrap();
4158 assert_eq!(gi.triple_count(), 1);
4159 assert!(rete.graph_index("http://ex/missing").is_none());
4160
4161 assert_eq!(rete.header().quad_count, 2);
4164
4165 assert_eq!(rete.query(Some("Alice"), None, None).len(), 1);
4167
4168 assert_eq!(
4170 rete.dump(None),
4171 vec![("Alice".into(), "knows".into(), "Bob".into())]
4172 );
4173 assert_eq!(
4174 rete.dump(Some("http://ex/g1")),
4175 vec![("Bob".into(), "age".into(), "30".into())]
4176 );
4177 }
4178
4179 #[test]
4180 fn query_in_graph_is_graph_scoped() {
4181 let mut db = DictionaryBuilder::new();
4184 for (s, p, o) in [
4185 ("Alice", "knows", "Bob"),
4186 ("Alice", "knows", "Carol"),
4187 ("Alice", "knows", "Dave"),
4188 ] {
4189 db.observe(s, p, o);
4190 }
4191 let dict = db.build();
4192
4193 let mut def = GraphIndexBuilder::new();
4194 def.push(dict.encode("Alice", "knows", "Bob").unwrap());
4195 def.push(dict.encode("Alice", "knows", "Carol").unwrap());
4196 let mut g1 = GraphIndexBuilder::new();
4197 g1.push(dict.encode("Alice", "knows", "Dave").unwrap());
4198
4199 let named = vec![("http://ex/g1".to_string(), g1.build())];
4200 let bytes = write_dataset(&dict, &def.build(), &named, true, &[], 0);
4201 let rete = Rete::open(&bytes).unwrap();
4202
4203 let mut def_objs: Vec<String> = rete
4205 .query_in_graph(None, Some("Alice"), Some("knows"), None)
4206 .into_iter()
4207 .map(|(_, _, o)| o)
4208 .collect();
4209 def_objs.sort();
4210 assert_eq!(def_objs, vec!["Bob".to_string(), "Carol".to_string()]);
4211
4212 assert_eq!(
4214 rete.query_in_graph(Some("http://ex/g1"), Some("Alice"), None, None),
4215 vec![("Alice".into(), "knows".into(), "Dave".into())]
4216 );
4217
4218 assert_eq!(rete.query_in_graph(None, None, None, None).len(), 2);
4220 assert_eq!(
4221 rete.query_in_graph(Some("http://ex/g1"), None, None, None)
4222 .len(),
4223 1
4224 );
4225
4226 assert!(rete
4228 .query_in_graph(Some("http://ex/missing"), None, None, None)
4229 .is_empty());
4230 }
4231
4232 #[test]
4233 fn query_quads_tags_every_graph() {
4234 let mut db = DictionaryBuilder::new();
4235 for (s, p, o) in [("Alice", "knows", "Bob"), ("Alice", "knows", "Dave")] {
4236 db.observe(s, p, o);
4237 }
4238 let dict = db.build();
4239 let mut def = GraphIndexBuilder::new();
4240 def.push(dict.encode("Alice", "knows", "Bob").unwrap());
4241 let mut g1 = GraphIndexBuilder::new();
4242 g1.push(dict.encode("Alice", "knows", "Dave").unwrap());
4243 let named = vec![("http://ex/g1".to_string(), g1.build())];
4244 let bytes = write_dataset(&dict, &def.build(), &named, true, &[], 0);
4245 let rete = Rete::open(&bytes).unwrap();
4246
4247 let quads = rete.query_quads(Some("Alice"), Some("knows"), None);
4249 assert_eq!(quads.len(), 2);
4250 assert_eq!(
4251 quads[0],
4252 (("Alice".into(), "knows".into(), "Bob".into()), None)
4253 );
4254 assert_eq!(
4255 quads[1],
4256 (
4257 ("Alice".into(), "knows".into(), "Dave".into()),
4258 Some("http://ex/g1".to_string())
4259 )
4260 );
4261
4262 assert!(rete.query_quads(Some("Nobody"), None, None).is_empty());
4264 }
4265
4266 #[test]
4267 fn pyramid_meta_round_trips_in_file() {
4268 let rete = Rete::open(&build_image()).unwrap();
4269 let pyr = rete.pyramid().expect("file has a pyramid");
4270 let total: u32 = pyr.summary.iter().map(|e| e.count).sum();
4272 assert_eq!(total, 3);
4273 assert!(!pyr.summary.is_empty());
4274 assert!(pyr.tiles.is_empty());
4275 }
4276
4277 #[test]
4278 fn schema_pyramid_round_trips_through_file_index_free() {
4279 use crate::reader::{CountingReader, SliceReader};
4280 let sub = "<http://www.w3.org/2000/01/rdf-schema#subClassOf>";
4281 let q = |s: &str, p: &str, o: &str| {
4282 (s.to_string(), p.to_string(), o.to_string(), None::<String>)
4283 };
4284 let quads = vec![
4286 q("<a>", RDF_TYPE, "<Astronomer>"),
4287 q("<b>", RDF_TYPE, "<Astronomer>"),
4288 q("<c>", RDF_TYPE, "<Person>"),
4289 q("<Astronomer>", sub, "<Scientist>"),
4290 q("<Scientist>", sub, "<Person>"),
4291 q("<Person>", sub, "<Agent>"),
4292 q("<a>", "<knows>", "<b>"),
4293 q("<b>", "<knows>", "<c>"),
4294 ];
4295 let (bytes, _) =
4296 crate::ingest::assemble_dataset_with_opts(quads, true, false, None, |_, _| Vec::new());
4297
4298 let rete = Rete::open(&bytes).unwrap();
4300 let pyr = rete.pyramid().expect("pyramid present");
4301 assert!(!pyr.level_rollups.is_empty(), "schema pyramid shipped");
4302 assert!(pyr
4303 .class_hierarchy
4304 .iter()
4305 .any(|n| n.class == "<Agent>" && n.depth == 0));
4306
4307 let r = CountingReader::new(SliceReader::new(&bytes));
4309 let view = SummaryView::open_ranged(&r).unwrap().unwrap();
4310 assert!(view.level_count() >= 2, "multi-level pyramid");
4311 let coarse = view.level_rollup(0).unwrap();
4312 assert!(
4313 coarse.classes.iter().any(|(c, _)| c == "<Agent>"),
4314 "coarsest level rolls up to the root Agent"
4315 );
4316 let h = Header::from_bytes(&bytes).unwrap();
4317 assert!(
4318 r.bytes_read() <= bytes.len() as u64 - h.root_dir_len,
4319 "summary read {} bytes; the {}-byte index section must be skipped",
4320 r.bytes_read(),
4321 h.root_dir_len
4322 );
4323 }
4324
4325 #[test]
4326 fn predicate_totals_from_summary_only() {
4327 use crate::reader::SliceReader;
4328 let bytes = build_image();
4330 let reader = SliceReader::new(&bytes);
4331 let view = SummaryView::open_ranged(&reader).unwrap().unwrap();
4332 assert_eq!(view.predicate_total("knows"), 2);
4333 assert_eq!(view.predicate_total("age"), 1);
4334 assert_eq!(view.predicate_total("missing"), 0);
4335 let totals = view.predicate_totals();
4336 assert_eq!(totals[0], ("knows".to_string(), 2)); }
4338
4339 #[test]
4340 fn query_patterns_resolve_to_terms() {
4341 let rete = Rete::open(&build_image()).unwrap();
4342
4343 assert_eq!(rete.query(None, None, None).len(), 3);
4345
4346 let mut alice = rete.query(Some("Alice"), None, None);
4348 alice.sort();
4349 assert_eq!(
4350 alice,
4351 vec![
4352 ("Alice".into(), "age".into(), "30".into()),
4353 ("Alice".into(), "knows".into(), "Bob".into()),
4354 ]
4355 );
4356
4357 assert_eq!(rete.query(None, Some("knows"), None).len(), 2);
4359
4360 assert_eq!(
4362 rete.query(Some("Bob"), Some("knows"), Some("Carol")),
4363 vec![("Bob".into(), "knows".into(), "Carol".into())]
4364 );
4365 assert!(rete.query(Some("Nobody"), None, None).is_empty());
4366 assert!(rete.query(None, Some("likes"), None).is_empty());
4367 }
4368
4369 #[test]
4370 fn query_provenance_reports_terms_ids_sections_and_index_choice() {
4371 let bytes = build_image();
4372 let rete = Rete::open(&bytes).unwrap();
4373
4374 let mut matches = rete.query_with_provenance(None, Some("knows"), None);
4375 matches.sort_by(|a, b| a.terms.cmp(&b.terms));
4376
4377 assert_eq!(matches.len(), 2);
4378 assert_eq!(
4379 matches[0].terms,
4380 ("Alice".into(), "knows".into(), "Bob".into())
4381 );
4382 assert_eq!(
4383 matches[0].ids,
4384 rete.dictionary().encode("Alice", "knows", "Bob").unwrap()
4385 );
4386 assert_eq!(matches[0].graph.as_deref(), None);
4387 assert_eq!(
4388 matches[0].matched_pattern,
4389 (None, Some(matches[0].ids.1), None)
4390 );
4391 assert_eq!(
4392 matches[0].index_permutation,
4393 crate::index::IndexPermutation::Pos
4394 );
4395
4396 let h = rete.header();
4397 assert_eq!(matches[0].dictionary_range.offset, h.dictionary_offset);
4398 assert_eq!(matches[0].dictionary_range.len, h.dictionary_len);
4399 assert_eq!(matches[0].index_range.offset, h.root_dir_offset);
4400 assert_eq!(matches[0].index_range.len, h.root_dir_len);
4401 assert!(
4402 matches[0].index_section_range.offset > h.root_dir_offset,
4403 "POS is section 1, so its payload starts after the container header and SPO payload"
4404 );
4405 assert!(matches[0].index_section_range.len > 0);
4406 assert!(matches[0].index_section_range.end() <= matches[0].index_range.end());
4407 assert!(matches[0].index_section_range.len < matches[0].index_range.len);
4408 assert_eq!(
4409 matches[0].pyramid_range.as_ref().map(|r| (r.offset, r.len)),
4410 Some((h.pyramid_meta_offset, h.pyramid_meta_len))
4411 );
4412 let tile_range = matches[0].tile_range.expect("tiled file reports a tile");
4415 assert!(matches[0]
4416 .tile
4417 .as_deref()
4418 .unwrap()
4419 .starts_with(matches[0].index_permutation.name()));
4420 assert!(matches[0].index_section_range.offset <= tile_range.offset);
4421 assert!(tile_range.end() <= matches[0].index_section_range.end());
4422 }
4423
4424 fn build_labeled(n: usize) -> Vec<u8> {
4428 const LABEL: &str = "<http://www.w3.org/2000/01/rdf-schema#label>";
4429 const WORDS: &[&str] = &[
4430 "alanine",
4431 "benzene",
4432 "glucose",
4433 "dextrose",
4434 "ethanol",
4435 "formate",
4436 "heptane",
4437 "isoleucine",
4438 ];
4439 let triples: Vec<(String, String, String)> = (0..n)
4440 .flat_map(|i| {
4441 let s = format!("<http://ex/e{i}>");
4442 let w = WORDS[i % WORDS.len()];
4443 [
4444 (s.clone(), LABEL.to_string(), format!("\"{w}-{i:06}\"")),
4445 (
4446 s,
4447 "<http://ex/p>".to_string(),
4448 format!("<http://ex/c{}>", i % 64),
4449 ),
4450 ]
4451 })
4452 .collect();
4453 let mut db = DictionaryBuilder::new();
4454 for (s, p, o) in &triples {
4455 db.observe(s, p, o);
4456 }
4457 let dict = db.build();
4458 let ids: Vec<(u32, u32, u32)> = triples
4459 .iter()
4460 .map(|(s, p, o)| dict.encode(s, p, o).unwrap())
4461 .collect();
4462 let mut ib = GraphIndexBuilder::new();
4463 for &t in &ids {
4464 ib.push(t);
4465 }
4466 let (meta, levels) = build_pyramid_meta(&dict, &ids, DEFAULT_TILE_BUDGET);
4467 write_file(&dict, &ib.build(), false, &meta, levels)
4468 }
4469
4470 #[test]
4471 fn prefix_search_matches_a_filter_scan() {
4472 let bytes = build_labeled(800);
4475 let rete = Rete::open(&bytes).unwrap();
4476 let idx_subjects: std::collections::BTreeSet<String> = rete
4477 .prefix_search("glucose", 10_000)
4478 .into_iter()
4479 .map(|(_label, subject)| subject)
4480 .collect();
4481 let q = "SELECT ?s WHERE { ?s <http://www.w3.org/2000/01/rdf-schema#label> ?l \
4483 FILTER(STRSTARTS(LCASE(?l), \"glucose\")) }";
4484 let crate::QueryOutput::Select(_, rows) = crate::eval_query(&rete, q).unwrap() else {
4485 panic!("expected SELECT");
4486 };
4487 let scan_subjects: std::collections::BTreeSet<String> =
4488 rows.iter().map(|r| r.get("s").cloned().unwrap()).collect();
4489 assert_eq!(idx_subjects, scan_subjects, "index agrees with the scan");
4490 assert_eq!(
4491 idx_subjects.len(),
4492 100,
4493 "800/8 words = 100 glucose-* labels"
4494 );
4495 }
4496
4497 #[test]
4501 #[ignore]
4502 fn bench_prefix_search_vs_filter_scan() {
4503 use std::time::Instant;
4504 let n = 6000; let bytes = build_labeled(n);
4506 let rete = Rete::open(&bytes).unwrap();
4507 let q = "SELECT ?s WHERE { ?s <http://www.w3.org/2000/01/rdf-schema#label> ?l \
4508 FILTER(STRSTARTS(LCASE(?l), \"glucose\")) }";
4509 let reps = 200;
4510 let idx_n = rete.prefix_search("glucose", 100_000).len();
4511 let t = Instant::now();
4512 for _ in 0..reps {
4513 std::hint::black_box(rete.prefix_search("glucose", 100_000));
4514 }
4515 let idx_ms = t.elapsed().as_secs_f64() * 1000.0 / reps as f64;
4516 let t = Instant::now();
4517 for _ in 0..reps {
4518 let _ = std::hint::black_box(crate::eval_query(&rete, q).unwrap());
4519 }
4520 let scan_ms = t.elapsed().as_secs_f64() * 1000.0 / reps as f64;
4521 println!(
4522 "label prefix search over {n} labeled subjects ({idx_n} matches): \
4523 index {idx_ms:.4} ms vs FILTER scan {scan_ms:.3} ms ({:.0}× faster)",
4524 scan_ms / idx_ms
4525 );
4526 }
4527
4528 #[test]
4532 #[ignore]
4533 fn bench_text_search_vs_contains_scan() {
4534 use std::time::Instant;
4535 const LABEL: &str = "<http://www.w3.org/2000/01/rdf-schema#label>";
4536 const WORDS: &[&str] = &[
4537 "alanine",
4538 "benzene",
4539 "glucose",
4540 "dextrose",
4541 "ethanol",
4542 "formate",
4543 "heptane",
4544 "isoleucine",
4545 ];
4546 let n = 6000;
4547 let triples: Vec<(String, String, String)> = (0..n)
4548 .map(|i| {
4549 (
4550 format!("<http://ex/e{i}>"),
4551 LABEL.to_string(),
4552 format!("\"{} sample number {i:06}\"", WORDS[i % WORDS.len()]),
4553 )
4554 })
4555 .collect();
4556 let mut db = DictionaryBuilder::new();
4557 for (s, p, o) in &triples {
4558 db.observe(s, p, o);
4559 }
4560 let dict = db.build();
4561 let ids: Vec<(u32, u32, u32)> = triples
4562 .iter()
4563 .map(|(s, p, o)| dict.encode(s, p, o).unwrap())
4564 .collect();
4565 let mut ib = GraphIndexBuilder::new();
4566 for &t in &ids {
4567 ib.push(t);
4568 }
4569 let ti = compute_text_index(&dict, &ids);
4570 let bytes = write_dataset_with_metadata(&dict, &ib.build(), &[], false, &[], 0, &[], &ti);
4571 let rete = Rete::open(&bytes).unwrap();
4572
4573 let q = "SELECT ?s WHERE { ?s <http://www.w3.org/2000/01/rdf-schema#label> ?l \
4574 FILTER(CONTAINS(LCASE(?l), \"glucose\")) }";
4575 let reps = 200;
4576 let idx_n = rete.text_search(&["glucose"], None, 100_000).len();
4577 let t = Instant::now();
4578 for _ in 0..reps {
4579 std::hint::black_box(rete.text_search(&["glucose"], None, 100_000));
4580 }
4581 let idx_ms = t.elapsed().as_secs_f64() * 1000.0 / reps as f64;
4582 let t = Instant::now();
4583 for _ in 0..reps {
4584 let _ = std::hint::black_box(crate::eval_query(&rete, q).unwrap());
4585 }
4586 let scan_ms = t.elapsed().as_secs_f64() * 1000.0 / reps as f64;
4587 println!(
4588 "text search over {n} literals ({idx_n} matches): \
4589 index {idx_ms:.4} ms vs FILTER(CONTAINS) scan {scan_ms:.3} ms ({:.0}× faster)",
4590 scan_ms / idx_ms
4591 );
4592 }
4593
4594 #[test]
4598 #[ignore = "operational tool, driven by RETE_DEBUG_* env vars"]
4599 fn debug_bound_po_routing() {
4600 struct FR(std::fs::File);
4601 impl crate::RangeReader for FR {
4602 fn len(&self) -> u64 {
4603 self.0.metadata().map(|m| m.len()).unwrap_or(0)
4604 }
4605 fn read_at(&self, offset: u64, len: u64) -> std::io::Result<Vec<u8>> {
4606 use std::os::unix::fs::FileExt;
4607 let mut buf = vec![0u8; len as usize];
4608 self.0.read_exact_at(&mut buf, offset)?;
4609 Ok(buf)
4610 }
4611 }
4612 let path = std::env::var("RETE_DEBUG_FILE").expect("RETE_DEBUG_FILE");
4613 let p_iri = std::env::var("RETE_DEBUG_P").expect("RETE_DEBUG_P");
4614 let o_iri = std::env::var("RETE_DEBUG_O").expect("RETE_DEBUG_O");
4615 let rete =
4616 Rete::open_ranged_lazy(std::sync::Arc::new(FR(std::fs::File::open(&path).unwrap())))
4617 .unwrap();
4618 let pid = rete.dict.predicate_id(&p_iri).expect("p resolves");
4619 let oid = rete.dict.object_id(&o_iri).expect("o resolves");
4620 eprintln!("pid={pid} oid={oid}");
4621 let pattern = (None, Some(pid), Some(oid));
4622 let perm = GraphIndex::best_permutation(pattern);
4623 eprintln!("best_permutation = {}", perm.name());
4624 let si = perm.section_index();
4625 let tiles = &rete.index.sections[si];
4626 eprintln!("section {} tiles = {}", perm.name(), tiles.len());
4627 let [pa, pb, pc] = perm.order_pattern(pattern);
4628 eprintln!("permuted pattern pa={pa:?} pb={pb:?} pc={pc:?}");
4629 let (start, end) = rete.index.tile_span(si, pa);
4630 eprintln!("tile_span = [{start}, {end}) -> {} tiles", end - start);
4631 let mut admitted = 0usize;
4632 for (ti, t) in tiles.iter().enumerate().take(end).skip(start) {
4633 if t.syn_admits(pb, pc) {
4634 admitted += 1;
4635 if admitted <= 10 {
4636 let (lo, hi) = t.leading_range();
4637 eprintln!(" admit tile {ti}: a=[{lo},{hi}] syn={:?}", t.syn);
4638 }
4639 }
4640 }
4641 eprintln!("admitted {admitted} tile(s) by synopsis");
4642 let n = rete.index.scan_iter(pattern).count();
4643 eprintln!("scan_iter matches = {n}");
4644 let hi_res = rete.query(None, Some(&p_iri), Some(&o_iri));
4645 eprintln!("high-level query matches = {}", hi_res.len());
4646 }
4647}