1use std::collections::{BTreeMap, BTreeSet, HashMap};
33
34use crate::meta::{CharSet, ClassNode, LevelLinks, LevelRollup, PredStat, PyramidMeta};
35use crate::varint::{read_uvarint, write_uvarint};
36
37const VERSION: u8 = 1;
39const DIGEST_LEN: usize = 16;
41const MAX_CHAR_SETS: usize = 128;
43const MAX_TRACKED_SHAPES: usize = 1 << 20;
45
46pub(crate) const RDFS_SUBCLASS_OF: &str = "<http://www.w3.org/2000/01/rdf-schema#subClassOf>";
48pub(crate) const OWL_DISJOINT_WITH: &str = "<http://www.w3.org/2002/07/owl#disjointWith>";
50pub(crate) const OWL_EQUIVALENT_CLASS: &str = "<http://www.w3.org/2002/07/owl#equivalentClass>";
52
53pub(crate) const UNTYPED: u32 = u32::MAX;
55pub(crate) const LITERAL: u32 = u32::MAX - 1;
57
58#[derive(Debug, Clone, Default, PartialEq)]
61pub struct SchemaMeta {
62 pub type_predicate: Option<String>,
65 pub predicate_stats: Vec<PredStat>,
67 pub char_sets: Vec<CharSet>,
69 pub class_hierarchy: Vec<ClassNode>,
71 pub level_rollups: Vec<LevelRollup>,
73 pub level_links: Vec<LevelLinks>,
75 pub subclass_cycles: Vec<Vec<String>>,
77 pub disjoint_pairs: Vec<(String, String)>,
79 pub equivalent_pairs: Vec<(String, String)>,
81}
82
83impl SchemaMeta {
84 pub fn is_empty(&self) -> bool {
86 self.predicate_stats.is_empty()
87 && self.char_sets.is_empty()
88 && self.class_hierarchy.is_empty()
89 && self.level_rollups.is_empty()
90 && self.level_links.is_empty()
91 && self.subclass_cycles.is_empty()
92 && self.disjoint_pairs.is_empty()
93 && self.equivalent_pairs.is_empty()
94 }
95
96 pub fn from_pyramid(p: &PyramidMeta) -> Self {
98 SchemaMeta {
99 type_predicate: None,
100 predicate_stats: p.predicate_stats.clone(),
101 char_sets: p.char_sets.clone(),
102 class_hierarchy: p.class_hierarchy.clone(),
103 level_rollups: p.level_rollups.clone(),
104 level_links: p.level_links.clone(),
105 subclass_cycles: p.subclass_cycles.clone(),
106 disjoint_pairs: p.disjoint_pairs.clone(),
107 equivalent_pairs: p.equivalent_pairs.clone(),
108 }
109 }
110
111 pub fn encode(&self) -> Vec<u8> {
113 if self.is_empty() {
114 return Vec::new();
115 }
116 let pm = PyramidMeta::new(0, Vec::new(), &[])
117 .with_schema(
118 self.class_hierarchy.clone(),
119 self.level_rollups.clone(),
120 self.level_links.clone(),
121 Vec::new(),
122 self.subclass_cycles.clone(),
123 self.disjoint_pairs.clone(),
124 self.equivalent_pairs.clone(),
125 )
126 .with_predicate_stats(self.predicate_stats.clone())
127 .with_char_sets(self.char_sets.clone());
128 let mut body = vec![VERSION];
129 let tp = self.type_predicate.as_deref().unwrap_or("");
130 write_uvarint(&mut body, tp.len() as u64);
131 body.extend_from_slice(tp.as_bytes());
132 body.extend_from_slice(&pm.encode());
133 let mut out = Vec::with_capacity(DIGEST_LEN + body.len());
134 out.extend_from_slice(&blake3::hash(&body).as_bytes()[..DIGEST_LEN]);
135 out.extend_from_slice(&body);
136 out
137 }
138
139 pub fn decode(section: &[u8]) -> Result<Self, crate::file::FileError> {
141 use crate::file::FileError;
142 if !crate::label_index::digest_ok(section) {
143 return Err(FileError::Container("schema section: digest mismatch"));
144 }
145 let body = §ion[DIGEST_LEN..];
146 if body.first() != Some(&VERSION) {
147 return Err(FileError::Container("schema section: unknown version"));
148 }
149 let (len, n) =
150 read_uvarint(&body[1..]).ok_or(FileError::Container("schema section: truncated"))?;
151 let start = 1 + n;
152 let end = start
153 .checked_add(len as usize)
154 .filter(|&e| e <= body.len())
155 .ok_or(FileError::Container("schema section: truncated"))?;
156 let tp = std::str::from_utf8(&body[start..end])
157 .map_err(|_| FileError::Container("schema section: bad utf8"))?;
158 let pm = PyramidMeta::decode(&body[end..])
159 .map_err(|_| FileError::Container("schema section: malformed blocks"))?;
160 Ok(SchemaMeta {
161 type_predicate: (!tp.is_empty()).then(|| tp.to_string()),
162 predicate_stats: pm.predicate_stats,
163 char_sets: pm.char_sets,
164 class_hierarchy: pm.class_hierarchy,
165 level_rollups: pm.level_rollups,
166 level_links: pm.level_links,
167 subclass_cycles: pm.subclass_cycles,
168 disjoint_pairs: pm.disjoint_pairs,
169 equivalent_pairs: pm.equivalent_pairs,
170 })
171 }
172}
173
174#[derive(Debug, Clone, Copy, PartialEq, Eq)]
176pub(crate) enum Space {
177 Subject,
178 Object,
179 Predicate,
180}
181
182#[derive(Debug, Clone, Default)]
184pub(crate) struct SchemaIds {
185 pub type_pid: Option<u32>,
187 pub type_term: Option<String>,
188 pub subclass_pid: Option<u32>,
189 pub disjoint_pid: Option<u32>,
190 pub equivalent_pid: Option<u32>,
191 pub n_shared: u32,
193 pub shared_literals: u32,
195 pub objonly_literals: u32,
198}
199
200impl SchemaIds {
201 fn is_literal(&self, o: u32) -> bool {
202 o <= self.shared_literals
203 || (o > self.n_shared && o - self.n_shared <= self.objonly_literals)
204 }
205
206 pub(crate) fn type_term_for(type_override: Option<&str>) -> String {
208 type_override.unwrap_or(crate::RDF_TYPE).to_string()
209 }
210
211 pub(crate) fn from_dictionary(dict: &crate::Dictionary, type_override: Option<&str>) -> Self {
213 let type_term = Self::type_term_for(type_override);
214 let type_pid = dict.predicate_id(&type_term);
215 let n_shared = dict.shared_count();
216 let is_lit = |t: Option<String>| t.is_some_and(|t| t.starts_with('"'));
217 let prefix_len = |n: u32, term: &dyn Fn(u32) -> Option<String>| -> u32 {
219 let (mut lo, mut hi) = (0u32, n);
220 while lo < hi {
221 let mid = lo + (hi - lo) / 2;
222 if is_lit(term(mid + 1)) {
223 lo = mid + 1;
224 } else {
225 hi = mid;
226 }
227 }
228 lo
229 };
230 let shared_literals = prefix_len(n_shared, &|i| dict.object_term(i));
231 let objonly_literals = prefix_len(dict.object_only_count(), &|i| {
232 dict.object_term(n_shared + i)
233 });
234 SchemaIds {
235 type_pid,
236 type_term: type_pid.map(|_| type_term),
237 subclass_pid: dict.predicate_id(RDFS_SUBCLASS_OF),
238 disjoint_pid: dict.predicate_id(OWL_DISJOINT_WITH),
239 equivalent_pid: dict.predicate_id(OWL_EQUIVALENT_CLASS),
240 n_shared,
241 shared_literals,
242 objonly_literals,
243 }
244 }
245}
246
247pub(crate) type DeferredLink = (u32, u32, u32);
249
250pub(crate) struct SchemaAccumulator {
255 ids: SchemaIds,
256 pstats: HashMap<u32, [u64; 5]>,
259 cur_s: Option<u32>,
260 cur_p: Option<u32>,
261 run_sp: u64,
262 shape: Vec<u32>,
263 shapes: HashMap<Vec<u32>, u64>,
264 shapes_frozen: bool,
265 cur_po: Option<(u32, u32)>,
266 run_po: u64,
267 instance: HashMap<u32, u64>,
268 subclass: Vec<(u32, u32)>,
269 disjoint: Vec<(u32, u32)>,
270 equivalent: Vec<(u32, u32)>,
271 links: HashMap<(u32, u32, u32), u64>,
272}
273
274impl SchemaAccumulator {
275 pub(crate) fn new(ids: SchemaIds) -> Self {
276 SchemaAccumulator {
277 ids,
278 pstats: HashMap::new(),
279 cur_s: None,
280 cur_p: None,
281 run_sp: 0,
282 shape: Vec::new(),
283 shapes: HashMap::new(),
284 shapes_frozen: false,
285 cur_po: None,
286 run_po: 0,
287 instance: HashMap::new(),
288 subclass: Vec::new(),
289 disjoint: Vec::new(),
290 equivalent: Vec::new(),
291 links: HashMap::new(),
292 }
293 }
294
295 #[cfg(not(target_arch = "wasm32"))]
297 pub(crate) fn ids(&self) -> &SchemaIds {
298 &self.ids
299 }
300
301 fn end_sp(&mut self) {
302 if let Some(p) = self.cur_p.take() {
303 let e = self.pstats.entry(p).or_default();
304 e[3] = e[3].max(self.run_sp);
305 }
306 self.run_sp = 0;
307 }
308
309 fn end_subject(&mut self) {
310 self.end_sp();
311 if self.cur_s.take().is_none() {
312 return;
313 }
314 let shape = std::mem::take(&mut self.shape);
315 if let Some(c) = self.shapes.get_mut(&shape) {
316 *c += 1;
317 } else if !self.shapes_frozen {
318 self.shapes.insert(shape, 1);
319 if self.shapes.len() > MAX_TRACKED_SHAPES {
320 self.shapes.retain(|_, c| *c > 1);
323 if self.shapes.len() > MAX_TRACKED_SHAPES / 2 {
324 self.shapes_frozen = true;
325 }
326 }
327 }
328 }
329
330 pub(crate) fn observe_spo(
335 &mut self,
336 s: u32,
337 p: u32,
338 o: u32,
339 class_s: Option<u32>,
340 ) -> Option<DeferredLink> {
341 if self.cur_s != Some(s) {
342 self.end_subject();
343 self.cur_s = Some(s);
344 }
345 if self.cur_p != Some(p) {
346 self.end_sp();
347 self.cur_p = Some(p);
348 self.shape.push(p);
349 self.pstats.entry(p).or_default()[1] += 1;
350 }
351 self.run_sp += 1;
352 self.pstats.entry(p).or_default()[0] += 1;
353
354 let ids = &self.ids;
355 if Some(p) == ids.type_pid {
356 *self.instance.entry(o).or_default() += 1;
357 return None;
358 }
359 if Some(p) == ids.subclass_pid {
360 self.subclass.push((s, o));
361 return None;
362 }
363 if Some(p) == ids.disjoint_pid {
364 self.disjoint.push((s, o));
365 }
366 if Some(p) == ids.equivalent_pid {
367 self.equivalent.push((s, o));
368 }
369 ids.type_pid?;
370 let cs = class_s.unwrap_or(UNTYPED);
371 if ids.is_literal(o) {
372 self.add_link(cs, p, LITERAL);
373 None
374 } else if o > ids.n_shared {
375 self.add_link(cs, p, UNTYPED);
377 None
378 } else {
379 Some((o, cs, p))
380 }
381 }
382
383 pub(crate) fn add_link(&mut self, class_s: u32, p: u32, class_o: u32) {
385 *self.links.entry((class_s, p, class_o)).or_default() += 1;
386 }
387
388 pub(crate) fn finish_spo(&mut self) {
390 self.end_subject();
391 }
392
393 fn end_po(&mut self) {
394 if let Some((p, _)) = self.cur_po.take() {
395 let e = self.pstats.entry(p).or_default();
396 e[4] = e[4].max(self.run_po);
397 }
398 self.run_po = 0;
399 }
400
401 pub(crate) fn observe_pos(&mut self, p: u32, o: u32) {
403 if self.cur_po != Some((p, o)) {
404 self.end_po();
405 self.cur_po = Some((p, o));
406 self.pstats.entry(p).or_default()[2] += 1;
407 }
408 self.run_po += 1;
409 }
410
411 pub(crate) fn finish_pos(&mut self) {
413 self.end_po();
414 }
415
416 pub(crate) fn finalize(self, resolve: &dyn Fn(Space, u32) -> Option<String>) -> SchemaMeta {
418 let mut predicate_stats: Vec<PredStat> = self
419 .pstats
420 .iter()
421 .map(|(&predicate, e)| PredStat {
422 predicate,
423 count: e[0],
424 distinct_subjects: e[1],
425 distinct_objects: e[2],
426 max_objects_per_subject: e[3].min(u32::MAX as u64) as u32,
427 max_subjects_per_object: e[4].min(u32::MAX as u64) as u32,
428 })
429 .collect();
430 predicate_stats.sort_by_key(|p| p.predicate);
431
432 let mut char_sets: Vec<CharSet> = self
433 .shapes
434 .into_iter()
435 .map(|(predicates, subjects)| CharSet {
436 predicates,
437 subjects,
438 })
439 .collect();
440 char_sets.sort_by(|a, b| {
441 b.subjects
442 .cmp(&a.subjects)
443 .then_with(|| a.predicates.cmp(&b.predicates))
444 });
445 char_sets.truncate(MAX_CHAR_SETS);
446
447 let mut out = SchemaMeta {
448 type_predicate: None,
449 predicate_stats,
450 char_sets,
451 ..SchemaMeta::default()
452 };
453 if self.instance.is_empty() {
454 return out;
455 }
456 out.type_predicate = self.ids.type_term.clone();
457 let class = |c: u32| -> Option<String> {
458 match c {
459 UNTYPED => Some("(untyped)".to_string()),
460 LITERAL => Some("(literal)".to_string()),
461 _ => resolve(Space::Object, c),
462 }
463 };
464 let mut instance_counts: BTreeMap<String, u64> = BTreeMap::new();
465 for (&c, &n) in &self.instance {
466 if let Some(t) = resolve(Space::Object, c) {
467 *instance_counts.entry(t).or_default() += n;
468 }
469 }
470 let mut parents: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
471 let mut self_loops: BTreeSet<String> = BTreeSet::new();
472 for &(s, o) in &self.subclass {
473 if let (Some(child), Some(parent)) =
474 (resolve(Space::Subject, s), resolve(Space::Object, o))
475 {
476 if child != parent {
477 parents.entry(child).or_default().insert(parent);
478 } else {
479 self_loops.insert(child);
480 }
481 }
482 }
483 let pairs = |v: &[(u32, u32)]| -> Vec<(String, String)> {
484 let mut set: BTreeSet<(String, String)> = BTreeSet::new();
485 for &(s, o) in v {
486 if let (Some(a), Some(b)) = (resolve(Space::Subject, s), resolve(Space::Object, o))
487 {
488 if a != b {
489 set.insert(if a < b { (a, b) } else { (b, a) });
490 }
491 }
492 }
493 set.into_iter().collect()
494 };
495 let disjoint_pairs = pairs(&self.disjoint);
496 let equivalent_pairs = pairs(&self.equivalent);
497 let mut leaf_links: BTreeMap<(String, String, String), u64> = BTreeMap::new();
498 for (&(cs, p, co), &n) in &self.links {
499 if let (Some(a), Some(pr), Some(b)) =
500 (class(cs), resolve(Space::Predicate, p), class(co))
501 {
502 *leaf_links.entry((a, pr, b)).or_default() += n;
503 }
504 }
505 let core = crate::schema_pyramid::assemble_schema(
506 &instance_counts,
507 &parents,
508 &self_loops,
509 &leaf_links,
510 &|_, _| 0,
511 );
512 out.class_hierarchy = core.class_hierarchy;
513 out.level_rollups = core.level_rollups;
514 out.level_links = core.level_links;
515 out.subclass_cycles = core.subclass_cycles;
516 out.disjoint_pairs = disjoint_pairs;
517 out.equivalent_pairs = equivalent_pairs;
518 out
519 }
520}
521
522pub(crate) struct ClassCursor<I: Iterator<Item = (u32, u32)>> {
525 it: I,
526 cur: Option<(u32, u32)>,
527}
528
529impl<I: Iterator<Item = (u32, u32)>> ClassCursor<I> {
530 pub(crate) fn new(mut it: I) -> Self {
531 let cur = it.next();
532 ClassCursor { it, cur }
533 }
534
535 pub(crate) fn get(&mut self, s: u32) -> Option<u32> {
537 while let Some((cs, _)) = self.cur {
538 if cs >= s {
539 break;
540 }
541 self.cur = self.it.next();
542 }
543 match self.cur {
544 Some((cs, c)) if cs == s => Some(c),
545 _ => None,
546 }
547 }
548}
549
550pub(crate) fn compute_in_memory(
554 dict: &crate::Dictionary,
555 triples: &mut [(u32, u32, u32)],
556 type_override: Option<&str>,
557) -> SchemaMeta {
558 let ids = SchemaIds::from_dictionary(dict, type_override);
559 let mut reps: Vec<(u32, u32)> = Vec::new();
562 if let Some(tp) = ids.type_pid {
563 for &(s, p, o) in triples.iter() {
564 if p == tp && reps.last().map(|r| r.0) != Some(s) {
565 reps.push((s, o));
566 }
567 }
568 }
569 let mut acc = SchemaAccumulator::new(ids);
570 let mut cursor = ClassCursor::new(reps.iter().copied());
571 for &(s, p, o) in triples.iter() {
572 let cs = cursor.get(s);
573 if let Some((obj, cls_s, pred)) = acc.observe_spo(s, p, o, cs) {
574 let co = reps
575 .binary_search_by_key(&obj, |r| r.0)
576 .map(|i| reps[i].1)
577 .unwrap_or(UNTYPED);
578 acc.add_link(cls_s, pred, co);
579 }
580 }
581 acc.finish_spo();
582 sort_pos(triples);
583 for &(_s, p, o) in triples.iter() {
584 acc.observe_pos(p, o);
585 }
586 acc.finish_pos();
587 acc.finalize(&|space, id| match space {
588 Space::Subject => dict.subject_term(id),
589 Space::Object => dict.object_term(id),
590 Space::Predicate => dict.predicate_term(id),
591 })
592}
593
594#[cfg(feature = "parallel")]
595fn sort_pos(v: &mut [(u32, u32, u32)]) {
596 use rayon::slice::ParallelSliceMut;
597 v.par_sort_unstable_by_key(|&(s, p, o)| (p, o, s));
598}
599#[cfg(not(feature = "parallel"))]
600fn sort_pos(v: &mut [(u32, u32, u32)]) {
601 v.sort_unstable_by_key(|&(s, p, o)| (p, o, s));
602}
603
604pub(crate) fn sort_dedup_spo(v: &mut Vec<(u32, u32, u32)>) {
606 #[cfg(feature = "parallel")]
607 {
608 use rayon::slice::ParallelSliceMut;
609 v.par_sort_unstable();
610 }
611 #[cfg(not(feature = "parallel"))]
612 v.sort_unstable();
613 v.dedup();
614}
615
616#[cfg(test)]
617mod tests {
618 use super::*;
619 use crate::DictionaryBuilder;
620
621 fn build(triples: &[(&str, &str, &str)]) -> (crate::Dictionary, Vec<(u32, u32, u32)>) {
622 let mut db = DictionaryBuilder::new();
623 for (s, p, o) in triples {
624 db.observe(s, p, o);
625 }
626 let dict = db.build();
627 let mut ids: Vec<_> = triples
628 .iter()
629 .map(|(s, p, o)| dict.encode(s, p, o).unwrap())
630 .collect();
631 sort_dedup_spo(&mut ids);
632 (dict, ids)
633 }
634
635 const T: &str = "<http://www.w3.org/1999/02/22-rdf-syntax-ns#type>";
636
637 #[test]
638 fn computes_stats_shapes_and_schema_without_a_dendrogram() {
639 let (dict, mut ids) = build(&[
640 ("<a>", T, "<Person>"),
641 ("<a>", "<knows>", "<b>"),
642 ("<a>", "<name>", "\"A\""),
643 ("<b>", T, "<Person>"),
644 ("<b>", "<name>", "\"B\""),
645 ("<b>", "<name>", "\"B2\""),
646 ("<c>", T, "<Org>"),
647 ("<a>", "<worksFor>", "<c>"),
648 ("<Person>", RDFS_SUBCLASS_OF, "<Agent>"),
649 ("<Org>", RDFS_SUBCLASS_OF, "<Agent>"),
650 ("<a>", "<knows>", "<b>"), ]);
652 let m = compute_in_memory(&dict, &mut ids, None);
653 let name = dict.predicate_id("<name>").unwrap();
654 let st = m
655 .predicate_stats
656 .iter()
657 .find(|s| s.predicate == name)
658 .unwrap();
659 assert_eq!(
660 (st.count, st.distinct_subjects, st.distinct_objects),
661 (3, 2, 3)
662 );
663 assert_eq!(st.max_objects_per_subject, 2);
664 let knows = dict.predicate_id("<knows>").unwrap();
665 assert_eq!(
666 m.predicate_stats
667 .iter()
668 .find(|s| s.predicate == knows)
669 .unwrap()
670 .count,
671 1,
672 "deduplicated"
673 );
674 assert!(!m.char_sets.is_empty());
675 assert_eq!(m.type_predicate.as_deref(), Some(T));
676 assert!(m.class_hierarchy.iter().any(|n| n.class == "<Person>"
677 && n.parents == vec!["<Agent>".to_string()]
678 && n.depth == 1));
679 let top = &m.level_rollups[0];
681 assert_eq!(top.round, 0);
682 assert_eq!(top.classes, vec![("<Agent>".to_string(), 3)]);
683 let leaf = m.level_links.last().unwrap();
685 let has = |s: &str, p: &str, o: &str, n: u64| {
686 leaf.links
687 .iter()
688 .any(|l| l.s_class == s && l.predicate == p && l.o_class == o && l.count == n)
689 };
690 assert!(has("<Person>", "<knows>", "<Person>", 1));
691 assert!(has("<Person>", "<worksFor>", "<Org>", 1));
692 assert!(has("<Person>", "<name>", "(literal)", 3));
693 let bytes = m.encode();
695 assert_eq!(SchemaMeta::decode(&bytes).unwrap(), m);
696 let mut bad = bytes.clone();
697 bad[20] ^= 0xff;
698 assert!(SchemaMeta::decode(&bad).is_err(), "digest catches a flip");
699 }
700
701 #[test]
702 fn untyped_graph_still_gets_stats() {
703 let (dict, mut ids) = build(&[("<a>", "<p>", "<b>"), ("<b>", "<p>", "\"x\"")]);
704 let m = compute_in_memory(&dict, &mut ids, None);
705 assert_eq!(m.predicate_stats.len(), 1);
706 assert!(m.class_hierarchy.is_empty() && m.level_rollups.is_empty());
707 assert_eq!(m.type_predicate, None);
708 assert!(!m.encode().is_empty());
709 }
710
711 #[test]
712 fn literal_ranges_are_found_by_binary_search() {
713 let (dict, _) = build(&[
714 ("<a>", "<p>", "\"1\""),
715 ("<a>", "<p>", "\"2\""),
716 ("<a>", "<p>", "<x>"),
717 ("<a>", "<p>", "_:b"),
718 ("<x>", "<p>", "<a>"),
719 ]);
720 let ids = SchemaIds::from_dictionary(&dict, None);
721 let lits: Vec<u32> = (1..=dict.shared_count() + dict.object_only_count())
722 .filter(|&o| ids.is_literal(o))
723 .collect();
724 let expect: Vec<u32> = (1..=dict.shared_count() + dict.object_only_count())
725 .filter(|&o| dict.object_term(o).unwrap().starts_with('"'))
726 .collect();
727 assert_eq!(lits, expect);
728 assert_eq!(lits.len(), 2);
729 }
730}