1use std::collections::{BTreeMap, HashMap};
18
19use crate::file::Rete;
20use crate::index::{GraphIndex, Pattern, Tile};
21use crate::row::{Ctx, Row, Slots, Val};
22
23#[derive(Debug, Clone, PartialEq, Eq)]
25pub enum PatternTerm {
26 Var(String),
27 Const(String),
28}
29
30impl PatternTerm {
31 pub fn parse(token: &str) -> Self {
33 if let Some(name) = token.strip_prefix('?') {
34 PatternTerm::Var(name.to_string())
35 } else {
36 PatternTerm::Const(token.to_string())
37 }
38 }
39}
40
41#[derive(Debug, Clone)]
43pub struct TriplePattern {
44 pub s: PatternTerm,
45 pub p: PatternTerm,
46 pub o: PatternTerm,
47}
48
49pub type Binding = BTreeMap<String, String>;
51
52#[derive(Clone, Copy)]
63enum SlotTerm {
64 Var(usize),
65 Node(u32),
66 Pred(u32),
67}
68
69fn pred_tag(p: u32) -> i64 {
70 -(p as i64) - 1
71}
72
73pub(crate) fn collect_pattern_slots(patterns: &[TriplePattern], slots: &mut Slots) {
75 for p in patterns {
76 for t in [&p.s, &p.p, &p.o] {
77 if let PatternTerm::Var(v) = t {
78 slots.add(v);
79 }
80 }
81 }
82}
83
84fn lower(patterns: &[TriplePattern], ctx: &Ctx) -> Option<Vec<(SlotTerm, SlotTerm, SlotTerm)>> {
87 let dict = ctx.rete.dictionary();
88 let node = |t: &PatternTerm| -> Option<SlotTerm> {
89 match t {
90 PatternTerm::Var(v) => ctx.slots.slot(v).map(SlotTerm::Var),
91 PatternTerm::Const(c) => dict.node_of_term(c).map(SlotTerm::Node),
92 }
93 };
94 let pred = |t: &PatternTerm| -> Option<SlotTerm> {
95 match t {
96 PatternTerm::Var(v) => ctx.slots.slot(v).map(SlotTerm::Var),
97 PatternTerm::Const(c) => dict.predicate_id(c).map(SlotTerm::Pred),
98 }
99 };
100 let mut lowered = Vec::with_capacity(patterns.len());
101 for p in patterns {
102 lowered.push((node(&p.s)?, pred(&p.p)?, node(&p.o)?));
103 }
104 Some(lowered)
105}
106
107pub fn eval_bgp(rete: &Rete, patterns: &[TriplePattern]) -> Vec<Binding> {
110 let mut slots = Slots::new();
111 collect_pattern_slots(patterns, &mut slots);
112 let ctx = Ctx::new(rete, slots);
113 eval_bgp_rows(&ctx, rete.default_index(), patterns)
114 .into_iter()
115 .map(|row| row_to_binding(&ctx, &row))
116 .collect()
117}
118
119pub(crate) fn row_to_binding(ctx: &Ctx, row: &Row) -> Binding {
122 let mut b = Binding::new();
123 for (i, v) in row.iter().enumerate() {
124 if let Some(val) = v {
125 if let Some(t) = ctx.resolver.str_once(val) {
126 b.insert(ctx.slots.name(i).to_string(), t);
127 }
128 }
129 }
130 b
131}
132
133pub(crate) fn eval_bgp_rows(ctx: &Ctx, index: &GraphIndex, patterns: &[TriplePattern]) -> Vec<Row> {
136 if patterns.is_empty() {
138 return vec![ctx.slots.empty_row()];
139 }
140 let Some(lowered) = lower(patterns, ctx) else {
142 return Vec::new();
143 };
144
145 let order = selectivity_order(ctx, &lowered);
149 let mut rows: Vec<Row> = vec![ctx.slots.empty_row()];
150 let mut bound: Vec<usize> = Vec::new();
151 let mut merged: [bool; 2] = [false, false];
152
153 if order.len() >= 2 {
158 if let Some((rel, slots)) =
159 try_merge_join(ctx, index, &lowered[order[0]], &lowered[order[1]])
160 {
161 rows = rel;
162 bound = slots;
163 merged = [true, true];
164 }
165 }
166
167 for (k, &idx) in order.iter().enumerate() {
168 if k < 2 && merged[k] {
169 continue; }
171 let t = lowered[idx];
172 let shares_bound = pattern_slots(&t).iter().any(|s| bound.contains(s));
182 let do_probe = !bound.is_empty()
189 && shares_bound
190 && !rows.is_empty()
191 && if index.is_remote() {
192 rows.len() <= remote_probe_max(index)
193 && (rows.len() <= REMOTE_PROBE_MIN || !pattern_is_selective(&t))
194 } else {
195 rows.len() <= BGP_PROBE_THRESHOLD
196 };
197 if do_probe {
198 let taken = std::mem::take(&mut rows);
199 if index.is_remote() {
205 let pats: Vec<Pattern> = taken
206 .iter()
207 .filter_map(|base| {
208 match (
209 probe_subject(ctx, &t.0, base),
210 probe_predicate(ctx, &t.1, base),
211 probe_object(ctx, &t.2, base),
212 ) {
213 (Some(s), Some(p), Some(o)) => Some((s, p, o)),
214 _ => None,
215 }
216 })
217 .collect();
218 index.prefetch_probe_tiles(&pats);
219 }
220 let mut next: Vec<Row> = Vec::with_capacity(taken.len());
221 for base in taken {
222 next.extend(probe_rows(ctx, index, t, base));
223 }
224 rows = next;
225 } else {
226 let Some((rel, rel_slots)) = pattern_rows(ctx, index, &t) else {
227 return Vec::new();
228 };
229 rows = hash_join(rows, &bound, rel, &rel_slots);
230 }
231 for s in pattern_slots(&t) {
232 if !bound.contains(&s) {
233 bound.push(s);
234 }
235 }
236 if rows.is_empty() {
237 break;
238 }
239 }
240 rows
241}
242
243fn try_merge_join(
255 ctx: &Ctx,
256 index: &GraphIndex,
257 ta: &(SlotTerm, SlotTerm, SlotTerm),
258 tb: &(SlotTerm, SlotTerm, SlotTerm),
259) -> Option<(Vec<Row>, Vec<usize>)> {
260 let dict = ctx.rete.dictionary();
261 let sa = pattern_slots(ta);
262 let sb = pattern_slots(tb);
263 let shared: Vec<usize> = sa.iter().copied().filter(|s| sb.contains(s)).collect();
264 if shared.len() != 1 {
265 return None;
266 }
267 let v = shared[0];
268 let col_of = |t: &(SlotTerm, SlotTerm, SlotTerm)| -> Option<usize> {
269 [&t.0, &t.1, &t.2]
270 .iter()
271 .position(|x| matches!(x, SlotTerm::Var(i) if *i == v))
272 };
273 let ca = col_of(ta)?;
274 let cb = col_of(tb)?;
275 if ca != cb {
276 return None; }
278 let (lo, hi) = {
288 let (a, b) = (pattern_scan_bytes(index, ta), pattern_scan_bytes(index, tb));
289 (a.min(b), a.max(b))
290 };
291 if hi >= FAT_SCAN_BYTES && hi / 4 >= lo {
292 return None;
293 }
294 let lower = |t: &(SlotTerm, SlotTerm, SlotTerm)| -> Option<Pattern> {
295 Some((
296 const_subject(&t.0, dict)?,
297 const_predicate(&t.1)?,
298 const_object(&t.2, dict)?,
299 ))
300 };
301 let collect = |t: &(SlotTerm, SlotTerm, SlotTerm), col: usize| -> Option<Vec<(u32, Row)>> {
304 let pat = lower(t)?;
305 let mut out = Vec::new();
306 for tri in index.scan_iter_sorted_on(pat, col)? {
307 if let Some(r) = triple_row(ctx, t, tri) {
308 out.push(([tri.0, tri.1, tri.2][col], r));
309 }
310 }
311 Some(out)
312 };
313 let rows_a = collect(ta, ca)?;
314 let rows_b = collect(tb, cb)?;
315
316 let mut out: Vec<Row> = Vec::new();
317 let (mut i, mut j) = (0usize, 0usize);
318 while i < rows_a.len() && j < rows_b.len() {
319 match rows_a[i].0.cmp(&rows_b[j].0) {
320 std::cmp::Ordering::Less => i += 1,
321 std::cmp::Ordering::Greater => j += 1,
322 std::cmp::Ordering::Equal => {
323 let key = rows_a[i].0;
324 let (i0, j0) = (i, j);
325 while i < rows_a.len() && rows_a[i].0 == key {
326 i += 1;
327 }
328 while j < rows_b.len() && rows_b[j].0 == key {
329 j += 1;
330 }
331 for (_, ra) in &rows_a[i0..i] {
332 for (_, rb) in &rows_b[j0..j] {
333 let mut row = ra.clone();
336 let mut ok = true;
337 for (slot, val) in rb.iter().enumerate() {
338 if let Some(val) = val {
339 match &row[slot] {
340 Some(existing) if existing != val => {
341 ok = false;
342 break;
343 }
344 _ => row[slot] = Some(val.clone()),
345 }
346 }
347 }
348 if ok {
349 out.push(row);
350 }
351 }
352 }
353 }
354 }
355 }
356 let mut slots = pattern_slots(ta);
357 for s in pattern_slots(tb) {
358 if !slots.contains(&s) {
359 slots.push(s);
360 }
361 }
362 Some((out, slots))
363}
364
365const BGP_PROBE_THRESHOLD: usize = 512;
373
374const REMOTE_PROBE_MIN: usize = 8;
383const REMOTE_PROBE_MAX: usize = 1024;
384
385fn remote_probe_max(index: &GraphIndex) -> usize {
392 REMOTE_PROBE_MAX * index.read_concurrency().clamp(1, 16)
393}
394
395pub(crate) fn prefetch_plan_probes(ctx: &Ctx, index: &GraphIndex, plan: &ProbePlan, rows: &[Row]) {
403 if !index.is_remote() {
404 return;
405 }
406 let Some(t) = plan.pats.first() else {
407 return;
408 };
409 let mut pats: Vec<Pattern> = Vec::new();
410 for base in rows {
411 if let (Some(s), Some(p), Some(o)) = (
412 probe_subject(ctx, &t.0, base),
413 probe_predicate(ctx, &t.1, base),
414 probe_object(ctx, &t.2, base),
415 ) {
416 if s.is_some() || o.is_some() {
417 pats.push((s, p, o));
418 }
419 }
420 }
421 index.prefetch_probe_tiles(&pats);
422}
423
424pub(crate) fn prefetch_subject_probes(
430 ctx: &Ctx,
431 index: &GraphIndex,
432 patterns: &[TriplePattern],
433 sv: &str,
434 subject_ids: &[u32],
435) {
436 if !index.is_remote() {
437 return;
438 }
439 let Some(lowered) = lower(patterns, ctx) else {
440 return;
441 };
442 let Some(slot) = ctx.slots.slot(sv) else {
443 return;
444 };
445 let dict = ctx.rete.dictionary();
446 let mut pats: Vec<Pattern> = Vec::new();
447 for &sid in subject_ids {
448 let mut base = ctx.slots.empty_row();
449 base[slot] = Some(Val::Id(dict.subject_node(sid) as i64));
450 for t in &lowered {
451 if !matches!(t.0, SlotTerm::Var(i) if i == slot) {
452 continue;
453 }
454 if let (Some(s), Some(p), Some(o)) = (
455 probe_subject(ctx, &t.0, &base),
456 probe_predicate(ctx, &t.1, &base),
457 probe_object(ctx, &t.2, &base),
458 ) {
459 pats.push((s, p, o));
460 }
461 }
462 }
463 index.prefetch_probe_tiles(&pats);
464}
465
466fn pattern_is_selective(t: &(SlotTerm, SlotTerm, SlotTerm)) -> bool {
469 matches!(t.0, SlotTerm::Node(_)) || matches!(t.2, SlotTerm::Node(_))
470}
471
472fn pattern_slots(t: &(SlotTerm, SlotTerm, SlotTerm)) -> Vec<usize> {
474 let mut slots: Vec<usize> = Vec::new();
475 for term in [&t.0, &t.1, &t.2] {
476 if let SlotTerm::Var(i) = term {
477 if !slots.contains(i) {
478 slots.push(*i);
479 }
480 }
481 }
482 slots
483}
484
485fn triple_row(
489 ctx: &Ctx,
490 t: &(SlotTerm, SlotTerm, SlotTerm),
491 (s_id, p_id, o_id): (u32, u32, u32),
492) -> Option<Row> {
493 let dict = ctx.rete.dictionary();
494 let s_val = dict.subject_node(s_id) as i64;
495 let p_val = ctx.resolver.canon_id(pred_tag(p_id));
496 let o_val = dict.object_node(o_id) as i64;
497 let mut row = ctx.slots.empty_row();
498 for (term, val) in [(&t.0, s_val), (&t.1, p_val), (&t.2, o_val)] {
499 if let SlotTerm::Var(i) = term {
500 match row[*i] {
501 Some(Val::Id(existing)) if existing != val => return None,
502 Some(_) => {}
503 None => row[*i] = Some(Val::Id(val)),
504 }
505 }
506 }
507 Some(row)
508}
509
510fn scan_rows<'q>(
515 ctx: &'q Ctx<'q>,
516 index: &'q GraphIndex,
517 t: (SlotTerm, SlotTerm, SlotTerm),
518) -> Option<impl Iterator<Item = Row> + 'q> {
519 let dict = ctx.rete.dictionary();
520 let (sid, pid, oid) = (
521 const_subject(&t.0, dict)?,
522 const_predicate(&t.1)?,
523 const_object(&t.2, dict)?,
524 );
525 Some(
526 index
527 .scan_iter((sid, pid, oid))
528 .filter_map(move |triple| triple_row(ctx, &t, triple)),
529 )
530}
531
532fn pattern_rows(
535 ctx: &Ctx,
536 index: &GraphIndex,
537 t: &(SlotTerm, SlotTerm, SlotTerm),
538) -> Option<(Vec<Row>, Vec<usize>)> {
539 let slots = pattern_slots(t);
540 let mut rel: Vec<Row> = Vec::new();
541 let dict = ctx.rete.dictionary();
542 let (sid, pid, oid) = (
543 const_subject(&t.0, dict)?,
544 const_predicate(&t.1)?,
545 const_object(&t.2, dict)?,
546 );
547 for triple in index.scan_iter((sid, pid, oid)) {
550 if let Some(row) = triple_row(ctx, t, triple) {
551 rel.push(row);
552 }
553 }
554 Some((rel, slots))
555}
556
557fn pattern_estimates(
569 ctx: &Ctx,
570 lowered: &[(SlotTerm, SlotTerm, SlotTerm)],
571 seed: &std::collections::HashSet<usize>,
572) -> Option<Vec<f64>> {
573 let pyr = ctx.rete.pyramid_if_loaded()?;
576 let mut pred: std::collections::HashMap<u32, u64> = std::collections::HashMap::new();
577 for e in &pyr.summary {
578 *pred.entry(e.predicate).or_insert(0) += e.count as u64;
579 }
580 let stats: std::collections::HashMap<u32, &crate::meta::PredStat> = pyr
583 .predicate_stats
584 .iter()
585 .map(|s| (s.predicate, s))
586 .collect();
587 let total = ctx.rete.header().quad_count.max(1) as f64;
588 let num_preds = pred.len().max(1) as f64;
589 const SEL_SUBJECT: f64 = 0.001;
591 const SEL_OBJECT: f64 = 0.02;
592 let node_bound = |t: &SlotTerm| match t {
593 SlotTerm::Node(_) => true,
594 SlotTerm::Var(v) => seed.contains(v),
595 SlotTerm::Pred(_) => false,
596 };
597 Some(
598 lowered
599 .iter()
600 .map(|t| {
601 let (base, st) = match t.1 {
606 SlotTerm::Pred(p) => {
607 (*pred.get(&p).unwrap_or(&0) as f64, stats.get(&p).copied())
608 }
609 SlotTerm::Var(v) if seed.contains(&v) => (total / num_preds, None),
610 _ => (total, None),
611 };
612 let mut est = base.max(1.0);
613 if node_bound(&t.0) {
614 est *= match st {
615 Some(s) if s.distinct_subjects > 0 => 1.0 / s.distinct_subjects as f64,
616 _ => SEL_SUBJECT,
617 };
618 }
619 if node_bound(&t.2) {
620 est *= match st {
621 Some(s) if s.distinct_objects > 0 => 1.0 / s.distinct_objects as f64,
622 _ => SEL_OBJECT,
623 };
624 }
625 est.max(1.0)
626 })
627 .collect(),
628 )
629}
630
631pub(crate) fn bgp_min_scan_bytes(
638 ctx: &Ctx,
639 index: &GraphIndex,
640 patterns: &[TriplePattern],
641) -> Option<u64> {
642 let lowered = lower(patterns, ctx)?;
643 lowered.iter().map(|t| pattern_scan_bytes(index, t)).min()
644}
645
646fn pattern_scan_bytes(index: &GraphIndex, t: &(SlotTerm, SlotTerm, SlotTerm)) -> u64 {
653 let sections = index.tile_sections();
654 let comp = |role: usize| match role {
655 0 => &t.0,
656 1 => &t.1,
657 _ => &t.2,
658 };
659 let mut best = u64::MAX;
660 for perm in crate::index::ALL_PERMS {
661 let tiles = sections[perm.section_index()];
662 best = best.min(match comp(perm.roles()[0]) {
663 SlotTerm::Node(id) | SlotTerm::Pred(id) => tiles
664 .iter()
665 .filter(|tile| {
666 let (lo, hi) = tile.leading_range();
667 lo <= *id && *id <= hi
668 })
669 .map(Tile::encoded_len)
670 .sum(),
671 SlotTerm::Var(_) => tiles.iter().map(Tile::encoded_len).sum(),
672 });
673 }
674 best
675}
676
677pub(crate) const FAT_SCAN_BYTES: u64 = 2 << 20;
683
684fn selectivity_order(ctx: &Ctx, lowered: &[(SlotTerm, SlotTerm, SlotTerm)]) -> Vec<usize> {
691 selectivity_order_seeded(ctx, lowered, &std::collections::HashSet::new())
692}
693
694fn selectivity_order_seeded(
698 ctx: &Ctx,
699 lowered: &[(SlotTerm, SlotTerm, SlotTerm)],
700 seed: &std::collections::HashSet<usize>,
701) -> Vec<usize> {
702 let estimates = pattern_estimates(ctx, lowered, seed);
703 let consts = |t: &(SlotTerm, SlotTerm, SlotTerm)| {
704 [&t.0, &t.1, &t.2]
705 .into_iter()
706 .filter(|x| match x {
707 SlotTerm::Var(v) => seed.contains(v),
708 _ => true,
709 })
710 .count()
711 };
712 let vars = |t: &(SlotTerm, SlotTerm, SlotTerm)| -> Vec<usize> {
713 [&t.0, &t.1, &t.2]
714 .into_iter()
715 .filter_map(|x| match x {
716 SlotTerm::Var(v) => Some(*v),
717 _ => None,
718 })
719 .collect()
720 };
721 let score = |i: usize| -> f64 {
725 match &estimates {
726 Some(e) => -e[i],
727 None => consts(&lowered[i]) as f64,
728 }
729 };
730 let type_pid = ctx.rete.dictionary().predicate_id(crate::file::RDF_TYPE);
738 let is_class_enum = |i: usize, bound: &std::collections::HashSet<usize>| -> bool {
739 let t = &lowered[i];
740 type_pid.is_some_and(|tp| matches!(t.1, SlotTerm::Pred(p) if p == tp))
741 && matches!(t.2, SlotTerm::Node(_))
742 && matches!(t.0, SlotTerm::Var(v) if !bound.contains(&v))
743 };
744 let n = lowered.len();
745 let mut remaining: Vec<usize> = (0..n).collect();
746 let mut order: Vec<usize> = Vec::with_capacity(n);
747 let mut bound: std::collections::HashSet<usize> = seed.clone();
748 while !remaining.is_empty() {
749 let best = *remaining
752 .iter()
753 .max_by(|&&a, &&b| {
754 let connected = |i: usize| vars(&lowered[i]).iter().any(|v| bound.contains(v));
755 connected(a)
756 .cmp(&connected(b))
757 .then_with(|| is_class_enum(b, &bound).cmp(&is_class_enum(a, &bound)))
758 .then_with(|| {
759 score(a)
760 .partial_cmp(&score(b))
761 .unwrap_or(std::cmp::Ordering::Equal)
762 })
763 .then_with(|| b.cmp(&a))
764 })
765 .unwrap();
766 for v in vars(&lowered[best]) {
767 bound.insert(v);
768 }
769 order.push(best);
770 remaining.retain(|&i| i != best);
771 }
772 order
773}
774
775pub(crate) fn bgp_exists(ctx: &Ctx, index: &GraphIndex, patterns: &[TriplePattern]) -> bool {
781 let dict = ctx.rete.dictionary();
782 let Some(lowered) = lower(patterns, ctx) else {
783 return false;
784 };
785 if let [t] = lowered.as_slice() {
786 let names: Vec<usize> = [&t.0, &t.1, &t.2]
790 .into_iter()
791 .filter_map(|x| match x {
792 SlotTerm::Var(v) => Some(*v),
793 _ => None,
794 })
795 .collect();
796 let distinct = names
797 .iter()
798 .enumerate()
799 .all(|(i, v)| !names[i + 1..].contains(v));
800 if distinct {
801 return match (
802 const_subject(&t.0, dict),
803 const_predicate(&t.1),
804 const_object(&t.2, dict),
805 ) {
806 (Some(s), Some(p), Some(o)) => index.scan_iter((s, p, o)).next().is_some(),
807 _ => false,
808 };
809 }
810 }
811 !eval_bgp_rows(ctx, index, patterns).is_empty()
812}
813
814pub(crate) struct BgpSolutions<'q> {
825 scan: Option<Box<dyn Iterator<Item = Row> + 'q>>,
826 prefix: Vec<Row>,
827 buckets: HashMap<Vec<Val>, Vec<usize>>,
829 shared: Vec<usize>,
830 cartesian: bool,
831 seed_only: bool,
834 cur_scan: Option<Row>,
835 matches: Vec<usize>,
837 mi: usize,
838}
839
840impl<'q> BgpSolutions<'q> {
841 fn empty() -> Self {
843 BgpSolutions {
844 scan: None,
845 prefix: Vec::new(),
846 buckets: HashMap::new(),
847 shared: Vec::new(),
848 cartesian: false,
849 seed_only: false,
850 cur_scan: None,
851 matches: Vec::new(),
852 mi: 0,
853 }
854 }
855
856 pub(crate) fn new(ctx: &'q Ctx<'q>, index: &'q GraphIndex, patterns: &[TriplePattern]) -> Self {
857 if patterns.is_empty() {
859 return BgpSolutions {
860 scan: Some(Box::new(std::iter::once(ctx.slots.empty_row()))),
861 prefix: vec![ctx.slots.empty_row()],
862 buckets: HashMap::new(),
863 shared: Vec::new(),
864 cartesian: true,
865 seed_only: true,
866 cur_scan: None,
867 matches: Vec::new(),
868 mi: 0,
869 };
870 }
871 let Some(lowered) = lower(patterns, ctx) else {
873 return Self::empty();
874 };
875 let order = selectivity_order(ctx, &lowered);
878 let (&last_i, prefix_is) = order.split_last().unwrap();
879 let prefix_pats: Vec<TriplePattern> =
880 prefix_is.iter().map(|&i| patterns[i].clone()).collect();
881 let prefix = eval_bgp_rows(ctx, index, &prefix_pats);
882 if prefix.is_empty() {
883 return Self::empty();
884 }
885
886 let Some(scan) = scan_rows(ctx, index, lowered[last_i]) else {
888 return Self::empty();
889 };
890
891 let shared: Vec<usize> = pattern_slots(&lowered[last_i])
895 .into_iter()
896 .filter(|&s| prefix[0][s].is_some())
897 .collect();
898 let cartesian = shared.is_empty();
899 let seed_only = prefix.len() == 1 && prefix[0].iter().all(Option::is_none);
902 let mut buckets: HashMap<Vec<Val>, Vec<usize>> = HashMap::new();
903 if !cartesian {
904 for (i, r) in prefix.iter().enumerate() {
905 let key: Vec<Val> = shared.iter().map(|&s| r[s].clone().unwrap()).collect();
906 buckets.entry(key).or_default().push(i);
907 }
908 }
909 BgpSolutions {
910 scan: Some(Box::new(scan)),
911 prefix,
912 buckets,
913 shared,
914 cartesian,
915 seed_only,
916 cur_scan: None,
917 matches: Vec::new(),
918 mi: 0,
919 }
920 }
921}
922
923impl Iterator for BgpSolutions<'_> {
924 type Item = Row;
925
926 fn next(&mut self) -> Option<Row> {
927 if self.seed_only {
929 return self.scan.as_mut()?.next();
930 }
931 loop {
932 if self.mi < self.matches.len() {
934 let pi = self.matches[self.mi];
935 self.mi += 1;
936 let mut merged = self.prefix[pi].clone();
937 for (slot, v) in self.cur_scan.as_ref().unwrap().iter().enumerate() {
938 if v.is_some() {
939 merged[slot] = v.clone();
940 }
941 }
942 return Some(merged);
943 }
944 let s = self.scan.as_mut()?.next()?;
947 self.matches.clear();
948 if self.cartesian {
949 self.matches.extend(0..self.prefix.len());
950 } else {
951 let key: Vec<Val> = self.shared.iter().map(|&i| s[i].clone().unwrap()).collect();
952 if let Some(c) = self.buckets.get(&key) {
953 self.matches.extend_from_slice(c);
954 }
955 }
956 self.mi = 0;
957 self.cur_scan = Some(s);
958 }
959 }
960}
961
962fn probe_subject(ctx: &Ctx, t: &SlotTerm, base: &Row) -> Option<Option<u32>> {
976 let dict = ctx.rete.dictionary();
977 match t {
978 SlotTerm::Node(n) => dict.node_as_subject_id(*n).map(Some),
979 SlotTerm::Pred(_) => None,
980 SlotTerm::Var(i) => match &base[*i] {
981 None => Some(None),
982 Some(Val::Id(v)) if *v >= 0 => dict.node_as_subject_id(*v as u32).map(Some),
983 Some(_) => None,
985 },
986 }
987}
988
989fn probe_object(ctx: &Ctx, t: &SlotTerm, base: &Row) -> Option<Option<u32>> {
991 let dict = ctx.rete.dictionary();
992 match t {
993 SlotTerm::Node(n) => dict.node_as_object_id(*n).map(Some),
994 SlotTerm::Pred(_) => None,
995 SlotTerm::Var(i) => match &base[*i] {
996 None => Some(None),
997 Some(Val::Id(v)) if *v >= 0 => dict.node_as_object_id(*v as u32).map(Some),
998 Some(_) => None,
999 },
1000 }
1001}
1002
1003fn probe_predicate(ctx: &Ctx, t: &SlotTerm, base: &Row) -> Option<Option<u32>> {
1005 let dict = ctx.rete.dictionary();
1006 match t {
1007 SlotTerm::Pred(p) => Some(Some(*p)),
1008 SlotTerm::Node(_) => None,
1009 SlotTerm::Var(i) => match &base[*i] {
1010 None => Some(None),
1011 Some(Val::Id(v)) if *v < 0 => Some(Some((-v - 1) as u32)),
1012 Some(Val::Id(v)) => ctx
1014 .resolver
1015 .term(*v)
1016 .and_then(|t| dict.predicate_id(&t))
1017 .map(Some),
1018 Some(Val::Str(_)) => None,
1019 },
1020 }
1021}
1022
1023fn probe_rows<'q>(
1027 ctx: &'q Ctx<'q>,
1028 index: &'q GraphIndex,
1029 t: (SlotTerm, SlotTerm, SlotTerm),
1030 base: Row,
1031) -> Box<dyn Iterator<Item = Row> + 'q> {
1032 let (Some(sid), Some(pid), Some(oid)) = (
1033 probe_subject(ctx, &t.0, &base),
1034 probe_predicate(ctx, &t.1, &base),
1035 probe_object(ctx, &t.2, &base),
1036 ) else {
1037 return Box::new(std::iter::empty());
1038 };
1039 let dict = ctx.rete.dictionary();
1040 Box::new(
1041 index
1042 .scan_iter((sid, pid, oid))
1043 .filter_map(move |(s_id, p_id, o_id)| {
1044 let s_val = dict.subject_node(s_id) as i64;
1045 let p_val = ctx.resolver.canon_id(pred_tag(p_id));
1046 let o_val = dict.object_node(o_id) as i64;
1047 let mut row = base.clone();
1048 for (term, val) in [(&t.0, s_val), (&t.1, p_val), (&t.2, o_val)] {
1049 if let SlotTerm::Var(i) = term {
1050 match row[*i] {
1051 Some(Val::Id(existing)) if existing != val => return None,
1052 Some(Val::Id(_)) => {}
1053 Some(Val::Str(_)) => return None,
1054 None => row[*i] = Some(Val::Id(val)),
1055 }
1056 }
1057 }
1058 Some(row)
1059 }),
1060 )
1061}
1062
1063pub(crate) struct ProbePlan {
1065 pats: Vec<(SlotTerm, SlotTerm, SlotTerm)>,
1066}
1067
1068impl ProbePlan {
1069 pub(crate) fn new(ctx: &Ctx, patterns: &[TriplePattern], seed_mask: &[bool]) -> Option<Self> {
1073 let lowered = lower(patterns, ctx)?;
1074 let seed: std::collections::HashSet<usize> = seed_mask
1075 .iter()
1076 .enumerate()
1077 .filter_map(|(i, b)| b.then_some(i))
1078 .collect();
1079 let order = selectivity_order_seeded(ctx, &lowered, &seed);
1080 Some(ProbePlan {
1081 pats: order.into_iter().map(|i| lowered[i]).collect(),
1082 })
1083 }
1084}
1085
1086pub(crate) struct ProbeJoin<'q> {
1089 ctx: &'q Ctx<'q>,
1090 index: &'q GraphIndex,
1091 pats: Vec<(SlotTerm, SlotTerm, SlotTerm)>,
1092 stack: Vec<Box<dyn Iterator<Item = Row> + 'q>>,
1093}
1094
1095impl<'q> ProbeJoin<'q> {
1096 pub(crate) fn new(
1099 ctx: &'q Ctx<'q>,
1100 index: &'q GraphIndex,
1101 patterns: &[TriplePattern],
1102 ) -> Option<Self> {
1103 let plan = ProbePlan::new(ctx, patterns, &vec![false; ctx.slots.len()])?;
1104 Some(Self::from_plan(ctx, index, &plan, ctx.slots.empty_row()))
1105 }
1106
1107 pub(crate) fn from_plan(
1109 ctx: &'q Ctx<'q>,
1110 index: &'q GraphIndex,
1111 plan: &ProbePlan,
1112 seed: Row,
1113 ) -> Self {
1114 let pats = plan.pats.clone();
1115 let first = probe_rows(ctx, index, pats[0], seed);
1116 ProbeJoin {
1117 ctx,
1118 index,
1119 pats,
1120 stack: vec![first],
1121 }
1122 }
1123}
1124
1125impl Iterator for ProbeJoin<'_> {
1126 type Item = Row;
1127
1128 fn next(&mut self) -> Option<Row> {
1129 loop {
1130 let depth = self.stack.len();
1131 match self.stack.last_mut()?.next() {
1132 Some(row) => {
1133 if depth == self.pats.len() {
1134 return Some(row);
1135 }
1136 let it = probe_rows(self.ctx, self.index, self.pats[depth], row);
1137 self.stack.push(it);
1138 }
1139 None => {
1140 self.stack.pop();
1141 if self.stack.is_empty() {
1142 return None;
1143 }
1144 }
1145 }
1146 }
1147 }
1148}
1149
1150fn hash_join(
1154 left: Vec<Row>,
1155 left_bound: &[usize],
1156 right: Vec<Row>,
1157 right_slots: &[usize],
1158) -> Vec<Row> {
1159 if left.len() == 1 && left_bound.is_empty() {
1161 return right;
1162 }
1163 if left.is_empty() || right.is_empty() {
1164 return Vec::new();
1165 }
1166 let shared: Vec<usize> = right_slots
1167 .iter()
1168 .copied()
1169 .filter(|s| left_bound.contains(s))
1170 .collect();
1171 let fill = |l: &Row, r: &Row| -> Row {
1172 let mut out = l.clone();
1173 for &s in right_slots {
1174 out[s] = r[s].clone();
1175 }
1176 out
1177 };
1178 if shared.is_empty() {
1179 let mut out = Vec::with_capacity(left.len() * right.len());
1183 for l in &left {
1184 for r in &right {
1185 out.push(fill(l, r));
1186 }
1187 }
1188 return out;
1189 }
1190 let key_of = |b: &Row| -> Vec<Val> { shared.iter().map(|&s| b[s].clone().unwrap()).collect() };
1191 let mut out = Vec::new();
1198 if right.len() <= left.len() {
1199 let mut buckets: HashMap<Vec<Val>, Vec<Row>> = HashMap::new();
1200 for r in right {
1201 buckets.entry(key_of(&r)).or_default().push(r);
1202 }
1203 for l in &left {
1204 if let Some(rs) = buckets.get(&key_of(l)) {
1205 for r in rs {
1206 out.push(fill(l, r));
1207 }
1208 }
1209 }
1210 } else {
1211 let mut buckets: HashMap<Vec<Val>, Vec<Row>> = HashMap::new();
1213 for l in left {
1214 buckets.entry(key_of(&l)).or_default().push(l);
1215 }
1216 for r in &right {
1217 if let Some(ls) = buckets.get(&key_of(r)) {
1218 for l in ls {
1219 let mut row = r.clone();
1220 for &s in left_bound {
1221 if let Some(v) = &l[s] {
1222 row[s] = Some(v.clone());
1223 }
1224 }
1225 out.push(row);
1226 }
1227 }
1228 }
1229 }
1230 out
1231}
1232
1233fn const_subject(t: &SlotTerm, d: &crate::Dictionary) -> Option<Option<u32>> {
1238 match t {
1239 SlotTerm::Node(n) => d.node_as_subject_id(*n).map(Some),
1240 SlotTerm::Pred(_) => None, SlotTerm::Var(_) => Some(None),
1242 }
1243}
1244
1245fn const_object(t: &SlotTerm, d: &crate::Dictionary) -> Option<Option<u32>> {
1246 match t {
1247 SlotTerm::Node(n) => d.node_as_object_id(*n).map(Some),
1248 SlotTerm::Pred(_) => None,
1249 SlotTerm::Var(_) => Some(None),
1250 }
1251}
1252
1253fn const_predicate(t: &SlotTerm) -> Option<Option<u32>> {
1254 match t {
1255 SlotTerm::Pred(p) => Some(Some(*p)),
1256 SlotTerm::Node(_) => None, SlotTerm::Var(_) => Some(None),
1258 }
1259}
1260
1261#[cfg(test)]
1262mod tests {
1263 use super::*;
1264 use crate::dictionary::DictionaryBuilder;
1265 use crate::file::write_file;
1266 use crate::index::GraphIndexBuilder;
1267
1268 fn rete_from(triples: &[(&str, &str, &str)]) -> Vec<u8> {
1269 let mut db = DictionaryBuilder::new();
1270 for (s, p, o) in triples {
1271 db.observe(s, p, o);
1272 }
1273 let dict = db.build();
1274 let mut ib = GraphIndexBuilder::new();
1275 for (s, p, o) in triples {
1276 ib.push(dict.encode(s, p, o).unwrap());
1277 }
1278 write_file(&dict, &ib.build(), false, &[], 0)
1279 }
1280
1281 fn pat(s: &str, p: &str, o: &str) -> TriplePattern {
1282 TriplePattern {
1283 s: PatternTerm::parse(s),
1284 p: PatternTerm::parse(p),
1285 o: PatternTerm::parse(o),
1286 }
1287 }
1288
1289 #[test]
1290 fn single_pattern_binds_variable() {
1291 let bytes = rete_from(&[("Alice", "knows", "Bob"), ("Bob", "knows", "Carol")]);
1292 let rete = Rete::open(&bytes).unwrap();
1293 let sols = eval_bgp(&rete, &[pat("Alice", "knows", "?y")]);
1294 assert_eq!(sols.len(), 1);
1295 assert_eq!(sols[0]["y"], "Bob");
1296 }
1297
1298 #[test]
1299 fn two_hop_join_on_shared_variable() {
1300 let bytes = rete_from(&[
1302 ("Alice", "knows", "Bob"),
1303 ("Bob", "knows", "Carol"),
1304 ("Carol", "knows", "Dave"),
1305 ("Alice", "knows", "Eve"), ]);
1307 let rete = Rete::open(&bytes).unwrap();
1308 let sols = eval_bgp(&rete, &[pat("?x", "knows", "?y"), pat("?y", "knows", "?z")]);
1309 let mut got: Vec<_> = sols
1311 .iter()
1312 .map(|b| (b["x"].clone(), b["y"].clone(), b["z"].clone()))
1313 .collect();
1314 got.sort();
1315 assert_eq!(
1316 got,
1317 vec![
1318 ("Alice".into(), "Bob".into(), "Carol".into()),
1319 ("Bob".into(), "Carol".into(), "Dave".into()),
1320 ]
1321 );
1322 }
1323
1324 #[test]
1325 fn repeated_variable_within_pattern() {
1326 let bytes = rete_from(&[("Alice", "knows", "Bob"), ("Bob", "knows", "Bob")]);
1328 let rete = Rete::open(&bytes).unwrap();
1329 let sols = eval_bgp(&rete, &[pat("?x", "knows", "?x")]);
1330 assert_eq!(sols.len(), 1);
1331 assert_eq!(sols[0]["x"], "Bob");
1332 }
1333
1334 #[test]
1335 fn no_solutions_yields_empty() {
1336 let bytes = rete_from(&[("Alice", "knows", "Bob")]);
1337 let rete = Rete::open(&bytes).unwrap();
1338 assert!(eval_bgp(&rete, &[pat("Alice", "likes", "?y")]).is_empty());
1339 }
1340
1341 #[test]
1348 fn snowflake_type_and_adjacency_matches_reference() {
1349 const TYPE: &str = "<http://www.w3.org/1999/02/22-rdf-syntax-ns#type>";
1350 let mut triples: Vec<(String, String, String)> = Vec::new();
1351 for i in 0..60 {
1353 let s = format!("S{i}");
1354 triples.push((s.clone(), TYPE.into(), "Student".into()));
1355 triples.push((s.clone(), "takesCourse".into(), format!("c{}", i % 5)));
1356 triples.push((s, "takesCourse".into(), format!("c{}", (i + 1) % 5)));
1357 }
1358 for j in 0..5 {
1360 triples.push((format!("c{j}"), TYPE.into(), "Course".into()));
1361 }
1362 triples.push(("Prof".into(), "teacherOf".into(), "c0".into()));
1363 triples.push(("Prof".into(), "teacherOf".into(), "c2".into()));
1364
1365 let refs: Vec<(&str, &str, &str)> = triples
1366 .iter()
1367 .map(|(s, p, o)| (s.as_str(), p.as_str(), o.as_str()))
1368 .collect();
1369 let bytes = rete_from(&refs);
1370 let rete = Rete::open(&bytes).unwrap();
1371
1372 let got: std::collections::BTreeSet<(String, String)> = eval_bgp(
1373 &rete,
1374 &[
1375 pat("?x", TYPE, "Student"),
1376 pat("?y", TYPE, "Course"),
1377 pat("?x", "takesCourse", "?y"),
1378 pat("Prof", "teacherOf", "?y"),
1379 ],
1380 )
1381 .into_iter()
1382 .map(|b| (b["x"].clone(), b["y"].clone()))
1383 .collect();
1384
1385 let taught = ["c0", "c2"];
1387 let mut want = std::collections::BTreeSet::new();
1388 for i in 0..60 {
1389 for c in [i % 5, (i + 1) % 5] {
1390 let course = format!("c{c}");
1391 if taught.contains(&course.as_str()) {
1392 want.insert((format!("S{i}"), course));
1393 }
1394 }
1395 }
1396 assert_eq!(got, want, "snowflake join must match the brute-force set");
1397 assert!(!want.is_empty(), "sanity: the reference set is non-empty");
1398 }
1399
1400 #[test]
1405 fn merge_join_subject_star_cross_product() {
1406 let bytes = rete_from(&[
1407 ("x", "a", "a1"),
1408 ("x", "a", "a2"),
1409 ("x", "b", "b1"),
1410 ("x", "b", "b2"),
1411 ("y", "a", "a3"),
1412 ("y", "b", "b3"),
1413 ("z", "a", "a4"), ]);
1415 let rete = Rete::open(&bytes).unwrap();
1416 let mut got: Vec<(String, String, String)> =
1417 eval_bgp(&rete, &[pat("?x", "a", "?av"), pat("?x", "b", "?bv")])
1418 .iter()
1419 .map(|m| (m["x"].clone(), m["av"].clone(), m["bv"].clone()))
1420 .collect();
1421 got.sort();
1422 let mut want: Vec<(String, String, String)> = [
1423 ("x", "a1", "b1"),
1424 ("x", "a1", "b2"),
1425 ("x", "a2", "b1"),
1426 ("x", "a2", "b2"),
1427 ("y", "a3", "b3"),
1428 ]
1429 .into_iter()
1430 .map(|(a, b, c)| (a.to_string(), b.to_string(), c.to_string()))
1431 .collect();
1432 want.sort();
1433 assert_eq!(got, want, "same-role merge must equal the brute-force join");
1434 }
1435}