1use std::collections::{BTreeMap, HashMap};
18
19use crate::file::Rete;
20use crate::index::{GraphIndex, Pattern};
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, index, &lowered);
149 let scan_est = tile_estimates(ctx, index, &lowered, &std::collections::HashSet::new());
153 let mut rows: Vec<Row> = vec![ctx.slots.empty_row()];
154 let mut bound: Vec<usize> = Vec::new();
155 let mut merged: [bool; 2] = [false, false];
156
157 if order.len() >= 2 {
162 if let Some((rel, slots)) =
163 try_merge_join(ctx, index, &lowered[order[0]], &lowered[order[1]])
164 {
165 rows = rel;
166 bound = slots;
167 merged = [true, true];
168 }
169 }
170
171 for (k, &idx) in order.iter().enumerate() {
172 if k < 2 && merged[k] {
173 continue; }
175 let t = lowered[idx];
176 let shares_bound = pattern_slots(&t).iter().any(|s| bound.contains(s));
186 let do_probe = !bound.is_empty()
197 && shares_bound
198 && !rows.is_empty()
199 && (probe_beats_scan(rows.len(), scan_est[idx])
200 || if index.is_remote() {
201 rows.len() <= remote_probe_max(index)
202 && (rows.len() <= REMOTE_PROBE_MIN || !pattern_is_selective(&t))
203 } else {
204 rows.len() <= BGP_PROBE_THRESHOLD
205 });
206 if do_probe {
207 let taken = std::mem::take(&mut rows);
208 if index.is_remote() {
214 let pats: Vec<Pattern> = taken
215 .iter()
216 .filter_map(|base| {
217 match (
218 probe_subject(ctx, &t.0, base),
219 probe_predicate(ctx, &t.1, base),
220 probe_object(ctx, &t.2, base),
221 ) {
222 (Some(s), Some(p), Some(o)) => Some((s, p, o)),
223 _ => None,
224 }
225 })
226 .collect();
227 index.prefetch_probe_tiles(&pats);
228 }
229 let mut next: Vec<Row> = Vec::with_capacity(taken.len());
230 for base in taken {
231 next.extend(probe_rows(ctx, index, t, base));
232 }
233 rows = next;
234 } else {
235 let Some((rel, rel_slots)) = pattern_rows(ctx, index, &t) else {
236 return Vec::new();
237 };
238 rows = hash_join(rows, &bound, rel, &rel_slots);
239 }
240 for s in pattern_slots(&t) {
241 if !bound.contains(&s) {
242 bound.push(s);
243 }
244 }
245 if rows.is_empty() {
246 break;
247 }
248 }
249 rows
250}
251
252fn try_merge_join(
264 ctx: &Ctx,
265 index: &GraphIndex,
266 ta: &(SlotTerm, SlotTerm, SlotTerm),
267 tb: &(SlotTerm, SlotTerm, SlotTerm),
268) -> Option<(Vec<Row>, Vec<usize>)> {
269 let dict = ctx.rete.dictionary();
270 let sa = pattern_slots(ta);
271 let sb = pattern_slots(tb);
272 let shared: Vec<usize> = sa.iter().copied().filter(|s| sb.contains(s)).collect();
273 if shared.len() != 1 {
274 return None;
275 }
276 let v = shared[0];
277 let col_of = |t: &(SlotTerm, SlotTerm, SlotTerm)| -> Option<usize> {
278 [&t.0, &t.1, &t.2]
279 .iter()
280 .position(|x| matches!(x, SlotTerm::Var(i) if *i == v))
281 };
282 let ca = col_of(ta)?;
283 let cb = col_of(tb)?;
284 if ca != cb {
285 return None; }
287 if merge_seed_lopsided(ctx, index, ta, tb) {
297 return None;
298 }
299 let lower = |t: &(SlotTerm, SlotTerm, SlotTerm)| -> Option<Pattern> {
300 Some((
301 const_subject(&t.0, dict)?,
302 const_predicate(&t.1)?,
303 const_object(&t.2, dict)?,
304 ))
305 };
306 let collect = |t: &(SlotTerm, SlotTerm, SlotTerm), col: usize| -> Option<Vec<(u32, Row)>> {
309 let pat = lower(t)?;
310 let mut out = Vec::new();
311 for tri in index.scan_iter_sorted_on(pat, col)? {
312 if let Some(r) = triple_row(ctx, t, tri) {
313 out.push(([tri.0, tri.1, tri.2][col], r));
314 }
315 }
316 Some(out)
317 };
318 let rows_a = collect(ta, ca)?;
319 let rows_b = collect(tb, cb)?;
320
321 let mut out: Vec<Row> = Vec::new();
322 let (mut i, mut j) = (0usize, 0usize);
323 while i < rows_a.len() && j < rows_b.len() {
324 match rows_a[i].0.cmp(&rows_b[j].0) {
325 std::cmp::Ordering::Less => i += 1,
326 std::cmp::Ordering::Greater => j += 1,
327 std::cmp::Ordering::Equal => {
328 let key = rows_a[i].0;
329 let (i0, j0) = (i, j);
330 while i < rows_a.len() && rows_a[i].0 == key {
331 i += 1;
332 }
333 while j < rows_b.len() && rows_b[j].0 == key {
334 j += 1;
335 }
336 for (_, ra) in &rows_a[i0..i] {
337 for (_, rb) in &rows_b[j0..j] {
338 let mut row = ra.clone();
341 let mut ok = true;
342 for (slot, val) in rb.iter().enumerate() {
343 if let Some(val) = val {
344 match &row[slot] {
345 Some(existing) if existing != val => {
346 ok = false;
347 break;
348 }
349 _ => row[slot] = Some(val.clone()),
350 }
351 }
352 }
353 if ok {
354 out.push(row);
355 }
356 }
357 }
358 }
359 }
360 }
361 let mut slots = pattern_slots(ta);
362 for s in pattern_slots(tb) {
363 if !slots.contains(&s) {
364 slots.push(s);
365 }
366 }
367 Some((out, slots))
368}
369
370const BGP_PROBE_THRESHOLD: usize = 512;
378
379const PROBE_SCAN_FLOOR: f64 = 65_536.0;
384
385const PROBE_RATIO: f64 = 16.0;
390
391fn probe_beats_scan(rows: usize, scan_est: f64) -> bool {
394 scan_est >= limits().probe_floor && (rows as f64) * PROBE_RATIO <= scan_est
395}
396
397pub(crate) fn probe_row_budget(est: f64) -> Option<usize> {
402 (est >= limits().probe_floor).then(|| (est / PROBE_RATIO) as usize)
403}
404
405pub(crate) fn bgp_min_estimate(
409 ctx: &Ctx,
410 index: &GraphIndex,
411 patterns: &[TriplePattern],
412) -> Option<f64> {
413 let lowered = lower(patterns, ctx)?;
414 tile_estimates(ctx, index, &lowered, &std::collections::HashSet::new())
415 .into_iter()
416 .reduce(f64::min)
417}
418
419pub(crate) fn bgp_total_estimate(
427 ctx: &Ctx,
428 index: &GraphIndex,
429 patterns: &[TriplePattern],
430) -> Option<f64> {
431 let lowered = lower(patterns, ctx)?;
432 Some(
433 tile_estimates(ctx, index, &lowered, &std::collections::HashSet::new())
434 .into_iter()
435 .sum(),
436 )
437}
438
439fn is_unbound_class(ctx: &Ctx, t: &(SlotTerm, SlotTerm, SlotTerm), shared: &[usize]) -> bool {
442 let type_pid = ctx.rete.dictionary().predicate_id(crate::file::RDF_TYPE);
443 matches!(
444 t,
445 (SlotTerm::Var(x), SlotTerm::Pred(p), SlotTerm::Var(c))
446 if Some(*p) == type_pid && shared.contains(x) && !shared.contains(c) && x != c
447 )
448}
449
450fn fat_enough(est: f64) -> bool {
454 #[cfg(test)]
455 if TEST_LIMITS.with(|c| c.get()).is_some() {
456 return est >= limits().probe_floor;
457 }
458 est >= FAT_SCAN_ROWS
459}
460
461pub(crate) struct CorrelatedProbe {
468 plan: Option<ProbePlan>,
470 slots: Vec<usize>,
472}
473
474impl CorrelatedProbe {
475 pub(crate) fn new(
479 ctx: &Ctx,
480 index: &GraphIndex,
481 patterns: &[TriplePattern],
482 seed: &[bool],
483 ) -> Option<Self> {
484 if patterns.is_empty() {
485 return None;
486 }
487 let mut slots: Vec<usize> = Vec::new();
488 for p in patterns {
489 for t in [&p.s, &p.p, &p.o] {
490 if let PatternTerm::Var(v) = t {
491 if let Some(i) = ctx.slots.slot(v) {
492 if !slots.contains(&i) {
493 slots.push(i);
494 }
495 }
496 }
497 }
498 }
499 if !slots.iter().any(|&i| seed[i]) {
500 return None;
501 }
502 let Some(est) = bgp_min_estimate(ctx, index, patterns) else {
504 return Some(CorrelatedProbe { plan: None, slots });
505 };
506 if !fat_enough(est) {
507 return None;
508 }
509 Some(CorrelatedProbe {
510 plan: ProbePlan::new(ctx, index, patterns, seed),
511 slots,
512 })
513 }
514
515 pub(crate) fn shares(&self, row: &Row) -> bool {
517 self.slots.iter().any(|&i| row[i].is_some())
518 }
519
520 pub(crate) fn probe(&self, ctx: &Ctx, index: &GraphIndex, row: &Row) -> bool {
524 match &self.plan {
525 None => false,
526 Some(plan) => ProbeJoin::from_plan(ctx, index, plan, row.clone())
527 .next()
528 .is_some(),
529 }
530 }
531}
532
533#[cfg(test)]
536pub(crate) fn with_probe_floor<T>(floor: f64, f: impl FnOnce() -> T) -> T {
537 let saved = TEST_LIMITS.with(|c| c.get());
538 TEST_LIMITS.with(|c| {
539 c.set(Some(PlanLimits {
540 probe_floor: floor,
541 ..limits()
542 }))
543 });
544 let out = f();
545 TEST_LIMITS.with(|c| c.set(saved));
546 out
547}
548
549#[derive(Clone, Copy)]
552struct PlanLimits {
553 probe_floor: f64,
554 stream_seed: f64,
555 hash_build: f64,
556}
557
558#[cfg(test)]
559thread_local! {
560 static TEST_LIMITS: std::cell::Cell<Option<PlanLimits>> = const { std::cell::Cell::new(None) };
561}
562
563fn limits() -> PlanLimits {
564 #[cfg(test)]
565 if let Some(l) = TEST_LIMITS.with(|c| c.get()) {
566 return l;
567 }
568 PlanLimits {
569 probe_floor: PROBE_SCAN_FLOOR,
570 stream_seed: STREAM_SEED_ROWS,
571 hash_build: HASH_BUILD_ROWS,
572 }
573}
574
575const STREAM_SEED_ROWS: f64 = (1u64 << 21) as f64;
583
584const HASH_BUILD_ROWS: f64 = (1u64 << 20) as f64;
587
588fn tile_estimates(
594 ctx: &Ctx,
595 index: &GraphIndex,
596 lowered: &[(SlotTerm, SlotTerm, SlotTerm)],
597 seed: &std::collections::HashSet<usize>,
598) -> Vec<f64> {
599 const SEL_SUBJECT: f64 = 0.001;
600 const SEL_OBJECT: f64 = 0.02;
601 const SEL_PREDICATE: f64 = 0.05;
602 let dict = ctx.rete.dictionary();
603 let total = ctx.rete.index_triples(index);
604 let seeded = |x: &SlotTerm| matches!(x, SlotTerm::Var(v) if seed.contains(v));
605 let needs_stats = lowered
613 .iter()
614 .any(|t| matches!(t.1, SlotTerm::Pred(_)) && (seeded(&t.0) || seeded(&t.2)));
615 let stats: std::collections::HashMap<u32, &crate::meta::PredStat> = if needs_stats {
616 ctx.rete
617 .planner_stats(true)
618 .iter()
619 .map(|s| (s.predicate, s))
620 .collect()
621 } else {
622 std::collections::HashMap::new()
623 };
624 lowered
625 .iter()
626 .map(|t| {
627 let (Some(s), Some(p), Some(o)) = (
628 const_subject(&t.0, dict),
629 const_predicate(&t.1),
630 const_object(&t.2, dict),
631 ) else {
632 return 0.0;
633 };
634 let mut est = index.estimate_matches((s, p, o), total);
635 if est == 0.0 {
636 return 0.0;
637 }
638 let st = match t.1 {
639 SlotTerm::Pred(pid) => stats.get(&pid).copied(),
640 _ => None,
641 };
642 if seeded(&t.0) {
643 est *= match st {
644 Some(st) if st.distinct_subjects > 0 => 1.0 / st.distinct_subjects as f64,
645 _ => SEL_SUBJECT,
646 };
647 }
648 if seeded(&t.1) {
649 est *= SEL_PREDICATE;
650 }
651 if seeded(&t.2) {
652 est *= match st {
653 Some(st) if st.distinct_objects > 0 => 1.0 / st.distinct_objects as f64,
654 _ => SEL_OBJECT,
655 };
656 }
657 est.max(1.0)
658 })
659 .collect()
660}
661
662const REMOTE_PROBE_MIN: usize = 8;
671const REMOTE_PROBE_MAX: usize = 1024;
672
673fn remote_probe_max(index: &GraphIndex) -> usize {
680 REMOTE_PROBE_MAX * index.read_concurrency().clamp(1, 16)
681}
682
683pub(crate) fn prefetch_plan_probes(ctx: &Ctx, index: &GraphIndex, plan: &ProbePlan, rows: &[Row]) {
691 if !index.is_remote() {
692 return;
693 }
694 let Some(t) = plan.pats.first() else {
695 return;
696 };
697 let mut pats: Vec<Pattern> = Vec::new();
698 for base in rows {
699 if let (Some(s), Some(p), Some(o)) = (
700 probe_subject(ctx, &t.0, base),
701 probe_predicate(ctx, &t.1, base),
702 probe_object(ctx, &t.2, base),
703 ) {
704 if s.is_some() || o.is_some() {
705 pats.push((s, p, o));
706 }
707 }
708 }
709 index.prefetch_probe_tiles(&pats);
710}
711
712pub(crate) fn prefetch_subject_probes(
718 ctx: &Ctx,
719 index: &GraphIndex,
720 patterns: &[TriplePattern],
721 sv: &str,
722 subject_ids: &[u32],
723) {
724 if !index.is_remote() {
725 return;
726 }
727 let Some(lowered) = lower(patterns, ctx) else {
728 return;
729 };
730 let Some(slot) = ctx.slots.slot(sv) else {
731 return;
732 };
733 let dict = ctx.rete.dictionary();
734 let mut pats: Vec<Pattern> = Vec::new();
735 for &sid in subject_ids {
736 let mut base = ctx.slots.empty_row();
737 base[slot] = Some(Val::Id(dict.subject_node(sid) as i64));
738 for t in &lowered {
739 if !matches!(t.0, SlotTerm::Var(i) if i == slot) {
740 continue;
741 }
742 if let (Some(s), Some(p), Some(o)) = (
743 probe_subject(ctx, &t.0, &base),
744 probe_predicate(ctx, &t.1, &base),
745 probe_object(ctx, &t.2, &base),
746 ) {
747 pats.push((s, p, o));
748 }
749 }
750 }
751 index.prefetch_probe_tiles(&pats);
752}
753
754fn pattern_is_selective(t: &(SlotTerm, SlotTerm, SlotTerm)) -> bool {
757 matches!(t.0, SlotTerm::Node(_)) || matches!(t.2, SlotTerm::Node(_))
758}
759
760fn pattern_slots(t: &(SlotTerm, SlotTerm, SlotTerm)) -> Vec<usize> {
762 let mut slots: Vec<usize> = Vec::new();
763 for term in [&t.0, &t.1, &t.2] {
764 if let SlotTerm::Var(i) = term {
765 if !slots.contains(i) {
766 slots.push(*i);
767 }
768 }
769 }
770 slots
771}
772
773fn pred_val(ctx: &Ctx, slot: &SlotTerm, p_id: u32) -> i64 {
780 match slot {
781 SlotTerm::Var(_) => ctx.resolver.canon_id(pred_tag(p_id)),
782 _ => pred_tag(p_id),
783 }
784}
785
786fn triple_row(
790 ctx: &Ctx,
791 t: &(SlotTerm, SlotTerm, SlotTerm),
792 (s_id, p_id, o_id): (u32, u32, u32),
793) -> Option<Row> {
794 crate::guard::check();
796 let dict = ctx.rete.dictionary();
797 let s_val = dict.subject_node(s_id) as i64;
798 let p_val = pred_val(ctx, &t.1, p_id);
799 let o_val = dict.object_node(o_id) as i64;
800 let mut row = ctx.slots.empty_row();
801 for (term, val) in [(&t.0, s_val), (&t.1, p_val), (&t.2, o_val)] {
802 if let SlotTerm::Var(i) = term {
803 match row[*i] {
804 Some(Val::Id(existing)) if existing != val => return None,
805 Some(_) => {}
806 None => row[*i] = Some(Val::Id(val)),
807 }
808 }
809 }
810 Some(row)
811}
812
813fn scan_rows<'q>(
818 ctx: &'q Ctx<'q>,
819 index: &'q GraphIndex,
820 t: (SlotTerm, SlotTerm, SlotTerm),
821) -> Option<impl Iterator<Item = Row> + 'q> {
822 let dict = ctx.rete.dictionary();
823 let (sid, pid, oid) = (
824 const_subject(&t.0, dict)?,
825 const_predicate(&t.1)?,
826 const_object(&t.2, dict)?,
827 );
828 Some(
829 index
830 .scan_iter((sid, pid, oid))
831 .filter_map(move |triple| triple_row(ctx, &t, triple)),
832 )
833}
834
835pub(crate) fn quad_scan_rows<'q>(
842 ctx: &'q Ctx<'q>,
843 patterns: &[TriplePattern],
844 gslot: usize,
845) -> Option<Box<dyn Iterator<Item = Row> + 'q>> {
846 let [p] = patterns else {
847 return None;
848 };
849 if [&p.s, &p.p, &p.o]
850 .iter()
851 .any(|t| matches!(t, PatternTerm::Var(v) if ctx.slots.slot(v) == Some(gslot)))
852 {
853 return None;
854 }
855 let dict = ctx.rete.dictionary();
856 let Some(lowered) = lower(patterns, ctx) else {
857 return Some(Box::new(std::iter::empty()));
858 };
859 let t = lowered[0];
860 let (Some(sid), Some(pid), Some(oid)) = (
861 const_subject(&t.0, dict),
862 const_predicate(&t.1),
863 const_object(&t.2, dict),
864 ) else {
865 return Some(Box::new(std::iter::empty()));
866 };
867 let scan = ctx.rete.named_quad_scan((sid, pid, oid))?;
868 Some(Box::new(scan.filter_map(move |(g, triple)| {
869 let mut row = triple_row(ctx, &t, triple)?;
870 row[gslot] = Some(Val::Id(dict.subject_node(g) as i64));
871 Some(row)
872 })))
873}
874
875pub(crate) type ScanPattern = (Option<u32>, Option<u32>, Option<u32>);
877
878pub(crate) fn driver_pattern(ctx: &Ctx, patterns: &[TriplePattern]) -> Option<ScanPattern> {
883 let dict = ctx.rete.dictionary();
884 let lowered = lower(patterns, ctx)?;
885 let mut best: Option<(u32, ScanPattern)> = None;
886 for t in &lowered {
887 let (sid, pid, oid) = (
888 const_subject(&t.0, dict)?,
889 const_predicate(&t.1)?,
890 const_object(&t.2, dict)?,
891 );
892 let score =
893 3 * u32::from(sid.is_some()) + u32::from(pid.is_some()) + 3 * u32::from(oid.is_some());
894 if score > 0 && best.is_none_or(|(b, _)| score > b) {
895 best = Some((score, (sid, pid, oid)));
896 }
897 }
898 best.map(|(_, pat)| pat)
899}
900
901fn pattern_rows(
904 ctx: &Ctx,
905 index: &GraphIndex,
906 t: &(SlotTerm, SlotTerm, SlotTerm),
907) -> Option<(Vec<Row>, Vec<usize>)> {
908 let slots = pattern_slots(t);
909 let mut rel: Vec<Row> = Vec::new();
910 let dict = ctx.rete.dictionary();
911 let (sid, pid, oid) = (
912 const_subject(&t.0, dict)?,
913 const_predicate(&t.1)?,
914 const_object(&t.2, dict)?,
915 );
916 for triple in index.scan_iter((sid, pid, oid)) {
919 if let Some(row) = triple_row(ctx, t, triple) {
920 rel.push(row);
921 }
922 }
923 Some((rel, slots))
924}
925
926fn pattern_estimates(
938 ctx: &Ctx,
939 index: &GraphIndex,
940 lowered: &[(SlotTerm, SlotTerm, SlotTerm)],
941 seed: &std::collections::HashSet<usize>,
942) -> Option<Vec<f64>> {
943 let pyr = ctx.rete.pyramid_if_loaded()?;
946 let mut pred: std::collections::HashMap<u32, u64> = std::collections::HashMap::new();
947 for e in &pyr.summary {
948 *pred.entry(e.predicate).or_insert(0) += e.count as u64;
949 }
950 let stats: std::collections::HashMap<u32, &crate::meta::PredStat> = ctx
957 .rete
958 .planner_stats(true)
959 .iter()
960 .map(|s| (s.predicate, s))
961 .collect();
962 let graph_total = ctx.rete.index_triples(index).max(1) as f64;
965 let total = graph_total;
966 let num_preds = pred.len().max(1) as f64;
967 const SEL_SUBJECT: f64 = 0.001;
969 const SEL_OBJECT: f64 = 0.02;
970 let node_bound = |t: &SlotTerm| match t {
971 SlotTerm::Node(_) => true,
972 SlotTerm::Var(v) => seed.contains(v),
973 SlotTerm::Pred(_) => false,
974 };
975 Some(
976 lowered
977 .iter()
978 .map(|t| {
979 let (base, st) = match t.1 {
984 SlotTerm::Pred(p) => {
985 (*pred.get(&p).unwrap_or(&0) as f64, stats.get(&p).copied())
986 }
987 SlotTerm::Var(v) if seed.contains(&v) => (total / num_preds, None),
988 _ => (total, None),
989 };
990 let mut est = base.min(graph_total).max(1.0);
991 if node_bound(&t.0) {
992 est *= match st {
993 Some(s) if s.distinct_subjects > 0 => 1.0 / s.distinct_subjects as f64,
994 _ => SEL_SUBJECT,
995 };
996 }
997 if node_bound(&t.2) {
998 est *= match st {
999 Some(s) if s.distinct_objects > 0 => 1.0 / s.distinct_objects as f64,
1000 _ => SEL_OBJECT,
1001 };
1002 }
1003 est.max(1.0)
1004 })
1005 .collect(),
1006 )
1007}
1008
1009pub(crate) fn bgp_min_scan_bytes(
1016 ctx: &Ctx,
1017 index: &GraphIndex,
1018 patterns: &[TriplePattern],
1019) -> Option<u64> {
1020 let lowered = lower(patterns, ctx)?;
1021 lowered
1022 .iter()
1023 .map(|t| pattern_scan_bytes(ctx, index, t))
1024 .min()
1025}
1026
1027fn pattern_scan_bytes(ctx: &Ctx, index: &GraphIndex, t: &(SlotTerm, SlotTerm, SlotTerm)) -> u64 {
1044 let dict = ctx.rete.dictionary();
1045 match (
1046 const_subject(&t.0, dict),
1047 const_predicate(&t.1),
1048 const_object(&t.2, dict),
1049 ) {
1050 (Some(s), Some(p), Some(o)) => routed_scan_bytes(index, (s, p, o)),
1051 _ => 0,
1052 }
1053}
1054
1055pub(crate) fn routed_scan_bytes(index: &GraphIndex, pattern: Pattern) -> u64 {
1057 let sections = index.tile_sections();
1058 let mut best = u64::MAX;
1059 for perm in index.perms().iter() {
1063 let tiles = sections[perm.section_index()];
1064 best = best.min(match perm.order_pattern(pattern)[0] {
1067 Some(id) => tiles.routed_bytes(id),
1068 None => tiles.total_bytes(),
1069 });
1070 }
1071 best
1072}
1073
1074pub(crate) const FAT_SCAN_ROWS: f64 = (1u64 << 20) as f64;
1087
1088fn merge_seed_lopsided(
1093 ctx: &Ctx,
1094 index: &GraphIndex,
1095 ta: &(SlotTerm, SlotTerm, SlotTerm),
1096 tb: &(SlotTerm, SlotTerm, SlotTerm),
1097) -> bool {
1098 let est = tile_estimates(ctx, index, &[*ta, *tb], &std::collections::HashSet::new());
1099 let (lo, hi) = (est[0].min(est[1]), est[0].max(est[1]));
1100 hi >= FAT_SCAN_ROWS && hi / 4.0 >= lo
1101}
1102
1103fn selectivity_order(
1110 ctx: &Ctx,
1111 index: &GraphIndex,
1112 lowered: &[(SlotTerm, SlotTerm, SlotTerm)],
1113) -> Vec<usize> {
1114 selectivity_order_seeded(ctx, index, lowered, &std::collections::HashSet::new())
1115}
1116
1117fn selectivity_order_seeded(
1121 ctx: &Ctx,
1122 index: &GraphIndex,
1123 lowered: &[(SlotTerm, SlotTerm, SlotTerm)],
1124 seed: &std::collections::HashSet<usize>,
1125) -> Vec<usize> {
1126 let (estimates, per_class) = match pattern_estimates(ctx, index, lowered, seed) {
1134 Some(e) => (Some(e), false),
1135 None => (Some(tile_estimates(ctx, index, lowered, seed)), true),
1136 };
1137 let consts = |t: &(SlotTerm, SlotTerm, SlotTerm)| {
1138 [&t.0, &t.1, &t.2]
1139 .into_iter()
1140 .filter(|x| match x {
1141 SlotTerm::Var(v) => seed.contains(v),
1142 _ => true,
1143 })
1144 .count()
1145 };
1146 let vars = |t: &(SlotTerm, SlotTerm, SlotTerm)| -> Vec<usize> {
1147 [&t.0, &t.1, &t.2]
1148 .into_iter()
1149 .filter_map(|x| match x {
1150 SlotTerm::Var(v) => Some(*v),
1151 _ => None,
1152 })
1153 .collect()
1154 };
1155 let score = |i: usize| -> f64 {
1159 match &estimates {
1160 Some(e) => -e[i],
1161 None => consts(&lowered[i]) as f64,
1162 }
1163 };
1164 let type_pid = ctx.rete.dictionary().predicate_id(crate::file::RDF_TYPE);
1172 let is_class_enum = |i: usize, bound: &std::collections::HashSet<usize>| -> bool {
1173 let t = &lowered[i];
1174 !per_class
1175 && type_pid.is_some_and(|tp| matches!(t.1, SlotTerm::Pred(p) if p == tp))
1176 && matches!(t.2, SlotTerm::Node(_))
1177 && matches!(t.0, SlotTerm::Var(v) if !bound.contains(&v))
1178 };
1179 let n = lowered.len();
1180 let mut remaining: Vec<usize> = (0..n).collect();
1181 let mut order: Vec<usize> = Vec::with_capacity(n);
1182 let mut bound: std::collections::HashSet<usize> = seed.clone();
1183 while !remaining.is_empty() {
1184 let best = *remaining
1187 .iter()
1188 .max_by(|&&a, &&b| {
1189 let connected = |i: usize| vars(&lowered[i]).iter().any(|v| bound.contains(v));
1190 connected(a)
1191 .cmp(&connected(b))
1192 .then_with(|| is_class_enum(b, &bound).cmp(&is_class_enum(a, &bound)))
1193 .then_with(|| {
1194 score(a)
1195 .partial_cmp(&score(b))
1196 .unwrap_or(std::cmp::Ordering::Equal)
1197 })
1198 .then_with(|| b.cmp(&a))
1199 })
1200 .unwrap();
1201 for v in vars(&lowered[best]) {
1202 bound.insert(v);
1203 }
1204 order.push(best);
1205 remaining.retain(|&i| i != best);
1206 }
1207 order
1208}
1209
1210pub(crate) fn bgp_exists(ctx: &Ctx, index: &GraphIndex, patterns: &[TriplePattern]) -> bool {
1216 let dict = ctx.rete.dictionary();
1217 let Some(lowered) = lower(patterns, ctx) else {
1218 return false;
1219 };
1220 if let [t] = lowered.as_slice() {
1221 let names: Vec<usize> = [&t.0, &t.1, &t.2]
1225 .into_iter()
1226 .filter_map(|x| match x {
1227 SlotTerm::Var(v) => Some(*v),
1228 _ => None,
1229 })
1230 .collect();
1231 let distinct = names
1232 .iter()
1233 .enumerate()
1234 .all(|(i, v)| !names[i + 1..].contains(v));
1235 if distinct {
1236 return match (
1237 const_subject(&t.0, dict),
1238 const_predicate(&t.1),
1239 const_object(&t.2, dict),
1240 ) {
1241 (Some(s), Some(p), Some(o)) => index.scan_iter((s, p, o)).next().is_some(),
1242 _ => false,
1243 };
1244 }
1245 }
1246 !eval_bgp_rows(ctx, index, patterns).is_empty()
1247}
1248
1249pub(crate) struct BgpSolutions<'q> {
1260 scan: Option<Box<dyn Iterator<Item = Row> + 'q>>,
1261 prefix: Vec<Row>,
1262 buckets: HashMap<Vec<Val>, Vec<usize>>,
1264 shared: Vec<usize>,
1265 cartesian: bool,
1266 seed_only: bool,
1269 cur_scan: Option<Row>,
1270 matches: Vec<usize>,
1272 mi: usize,
1273}
1274
1275impl<'q> BgpSolutions<'q> {
1276 fn passthrough(rows: Box<dyn Iterator<Item = Row> + 'q>) -> Self {
1278 BgpSolutions {
1279 scan: Some(rows),
1280 prefix: Vec::new(),
1281 buckets: HashMap::new(),
1282 shared: Vec::new(),
1283 cartesian: false,
1284 seed_only: true,
1285 cur_scan: None,
1286 matches: Vec::new(),
1287 mi: 0,
1288 }
1289 }
1290
1291 fn streaming(
1300 ctx: &'q Ctx<'q>,
1301 index: &'q GraphIndex,
1302 lowered: &[(SlotTerm, SlotTerm, SlotTerm)],
1303 order: &[usize],
1304 est: &[f64],
1305 ) -> Self {
1306 let (mut stream, mut bound, skip): (Box<dyn Iterator<Item = Row> + 'q>, Vec<usize>, usize) =
1307 match (est[order[1]] > limits().stream_seed)
1308 .then(|| merge_stream(ctx, index, lowered[order[0]], lowered[order[1]]))
1309 .flatten()
1310 {
1311 Some((m, slots)) => (m, slots, 2),
1312 None => {
1313 let t = lowered[order[0]];
1314 let Some(scan) = scan_rows(ctx, index, t) else {
1315 return Self::empty();
1316 };
1317 (Box::new(scan), pattern_slots(&t), 1)
1318 }
1319 };
1320 for &i in &order[skip..] {
1321 let t = lowered[i];
1322 let slots = pattern_slots(&t);
1323 let shared: Vec<usize> = slots
1324 .iter()
1325 .copied()
1326 .filter(|s| bound.contains(s))
1327 .collect();
1328 if !shared.is_empty() && est[i] > limits().hash_build {
1329 stream = Box::new(stream.flat_map(move |base| probe_rows(ctx, index, t, base)));
1330 } else {
1331 let Some((rel, _)) = pattern_rows(ctx, index, &t) else {
1334 return Self::empty();
1335 };
1336 let mut table: HashMap<Vec<Val>, Vec<Row>> = HashMap::new();
1337 for r in rel {
1338 let key: Vec<Val> = shared.iter().map(|&s| r[s].clone().unwrap()).collect();
1339 table.entry(key).or_default().push(r);
1340 }
1341 let fill = slots.clone();
1342 stream = Box::new(stream.flat_map(move |base| {
1343 let key: Vec<Val> = shared.iter().map(|&s| base[s].clone().unwrap()).collect();
1344 let out: Vec<Row> = table
1345 .get(&key)
1346 .map(|rs| {
1347 rs.iter()
1348 .map(|r| {
1349 let mut row = base.clone();
1350 for &s in &fill {
1351 row[s] = r[s].clone();
1352 }
1353 row
1354 })
1355 .collect()
1356 })
1357 .unwrap_or_default();
1358 out.into_iter()
1359 }));
1360 }
1361 for s in slots {
1362 if !bound.contains(&s) {
1363 bound.push(s);
1364 }
1365 }
1366 }
1367 Self::passthrough(stream)
1368 }
1369
1370 fn empty() -> Self {
1372 BgpSolutions {
1373 scan: None,
1374 prefix: Vec::new(),
1375 buckets: HashMap::new(),
1376 shared: Vec::new(),
1377 cartesian: false,
1378 seed_only: false,
1379 cur_scan: None,
1380 matches: Vec::new(),
1381 mi: 0,
1382 }
1383 }
1384
1385 pub(crate) fn new(ctx: &'q Ctx<'q>, index: &'q GraphIndex, patterns: &[TriplePattern]) -> Self {
1386 if patterns.is_empty() {
1388 return BgpSolutions {
1389 scan: Some(Box::new(std::iter::once(ctx.slots.empty_row()))),
1390 prefix: vec![ctx.slots.empty_row()],
1391 buckets: HashMap::new(),
1392 shared: Vec::new(),
1393 cartesian: true,
1394 seed_only: true,
1395 cur_scan: None,
1396 matches: Vec::new(),
1397 mi: 0,
1398 };
1399 }
1400 let Some(lowered) = lower(patterns, ctx) else {
1402 return Self::empty();
1403 };
1404 let order = selectivity_order(ctx, index, &lowered);
1407 let scan_est = tile_estimates(ctx, index, &lowered, &std::collections::HashSet::new());
1408 if order.len() >= 2 && scan_est[order[0]] > limits().stream_seed {
1411 return Self::streaming(ctx, index, &lowered, &order, &scan_est);
1412 }
1413 let (&last_i, prefix_is) = order.split_last().unwrap();
1414 let prefix_pats: Vec<TriplePattern> =
1415 prefix_is.iter().map(|&i| patterns[i].clone()).collect();
1416 let prefix = eval_bgp_rows(ctx, index, &prefix_pats);
1417 if prefix.is_empty() {
1418 return Self::empty();
1419 }
1420
1421 let shared: Vec<usize> = pattern_slots(&lowered[last_i])
1425 .into_iter()
1426 .filter(|&s| prefix[0][s].is_some())
1427 .collect();
1428
1429 let class_probe = !shared.is_empty()
1438 && scan_est[last_i] >= limits().probe_floor
1439 && (prefix.len() as f64) <= scan_est[last_i]
1440 && is_unbound_class(ctx, &lowered[last_i], &shared);
1441 if !shared.is_empty() && (probe_beats_scan(prefix.len(), scan_est[last_i]) || class_probe) {
1442 let t = lowered[last_i];
1443 if index.is_remote() {
1444 prefetch_rows_for(ctx, index, t, &prefix);
1445 }
1446 let stream = prefix
1447 .into_iter()
1448 .flat_map(move |base| probe_rows(ctx, index, t, base));
1449 return Self::passthrough(Box::new(stream));
1450 }
1451
1452 let Some(scan) = scan_rows(ctx, index, lowered[last_i]) else {
1454 return Self::empty();
1455 };
1456 let cartesian = shared.is_empty();
1457 let seed_only = prefix.len() == 1 && prefix[0].iter().all(Option::is_none);
1460 let mut buckets: HashMap<Vec<Val>, Vec<usize>> = HashMap::new();
1461 if !cartesian {
1462 for (i, r) in prefix.iter().enumerate() {
1463 let key: Vec<Val> = shared.iter().map(|&s| r[s].clone().unwrap()).collect();
1464 buckets.entry(key).or_default().push(i);
1465 }
1466 }
1467 BgpSolutions {
1468 scan: Some(Box::new(scan)),
1469 prefix,
1470 buckets,
1471 shared,
1472 cartesian,
1473 seed_only,
1474 cur_scan: None,
1475 matches: Vec::new(),
1476 mi: 0,
1477 }
1478 }
1479}
1480
1481impl Iterator for BgpSolutions<'_> {
1482 type Item = Row;
1483
1484 fn next(&mut self) -> Option<Row> {
1485 if self.seed_only {
1487 return self.scan.as_mut()?.next();
1488 }
1489 loop {
1490 if self.mi < self.matches.len() {
1492 let pi = self.matches[self.mi];
1493 self.mi += 1;
1494 let mut merged = self.prefix[pi].clone();
1495 for (slot, v) in self.cur_scan.as_ref().unwrap().iter().enumerate() {
1496 if v.is_some() {
1497 merged[slot] = v.clone();
1498 }
1499 }
1500 return Some(merged);
1501 }
1502 let s = self.scan.as_mut()?.next()?;
1505 self.matches.clear();
1506 if self.cartesian {
1507 self.matches.extend(0..self.prefix.len());
1508 } else {
1509 let key: Vec<Val> = self.shared.iter().map(|&i| s[i].clone().unwrap()).collect();
1510 if let Some(c) = self.buckets.get(&key) {
1511 self.matches.extend_from_slice(c);
1512 }
1513 }
1514 self.mi = 0;
1515 self.cur_scan = Some(s);
1516 }
1517 }
1518}
1519
1520fn prefetch_rows_for(
1523 ctx: &Ctx,
1524 index: &GraphIndex,
1525 t: (SlotTerm, SlotTerm, SlotTerm),
1526 rows: &[Row],
1527) {
1528 let pats: Vec<Pattern> = rows
1529 .iter()
1530 .filter_map(|base| {
1531 match (
1532 probe_subject(ctx, &t.0, base),
1533 probe_predicate(ctx, &t.1, base),
1534 probe_object(ctx, &t.2, base),
1535 ) {
1536 (Some(s), Some(p), Some(o)) => Some((s, p, o)),
1537 _ => None,
1538 }
1539 })
1540 .collect();
1541 index.prefetch_probe_tiles(&pats);
1542}
1543
1544#[allow(clippy::type_complexity)]
1552fn merge_stream<'q>(
1553 ctx: &'q Ctx<'q>,
1554 index: &'q GraphIndex,
1555 ta: (SlotTerm, SlotTerm, SlotTerm),
1556 tb: (SlotTerm, SlotTerm, SlotTerm),
1557) -> Option<(Box<dyn Iterator<Item = Row> + 'q>, Vec<usize>)> {
1558 let sa = pattern_slots(&ta);
1559 let sb = pattern_slots(&tb);
1560 let shared: Vec<usize> = sa.iter().copied().filter(|s| sb.contains(s)).collect();
1561 let [v] = shared[..] else {
1562 return None;
1563 };
1564 if !matches!(ta.0, SlotTerm::Var(i) if i == v) || !matches!(tb.0, SlotTerm::Var(i) if i == v) {
1565 return None;
1566 }
1567 let dict = ctx.rete.dictionary();
1568 let lower = |t: &(SlotTerm, SlotTerm, SlotTerm)| -> Option<Pattern> {
1569 Some((
1570 const_subject(&t.0, dict)?,
1571 const_predicate(&t.1)?,
1572 const_object(&t.2, dict)?,
1573 ))
1574 };
1575 let side = |t: (SlotTerm, SlotTerm, SlotTerm)| -> Option<KeyedRows<'q>> {
1576 let it = index.scan_iter_sorted_on(lower(&t)?, 0)?;
1577 let keyed: Box<dyn Iterator<Item = (u32, Row)> + 'q> =
1578 Box::new(it.filter_map(move |tri| triple_row(ctx, &t, tri).map(|r| (tri.0, r))));
1579 Some(keyed.peekable())
1580 };
1581 let a = side(ta)?;
1582 let b = side(tb)?;
1583 let mut slots = sa;
1584 for s in sb {
1585 if !slots.contains(&s) {
1586 slots.push(s);
1587 }
1588 }
1589 Some((
1590 Box::new(MergeStream {
1591 a,
1592 b,
1593 ra: Vec::new(),
1594 rb: Vec::new(),
1595 i: 0,
1596 j: 0,
1597 }),
1598 slots,
1599 ))
1600}
1601
1602type KeyedRows<'q> = std::iter::Peekable<Box<dyn Iterator<Item = (u32, Row)> + 'q>>;
1603
1604struct MergeStream<'q> {
1607 a: KeyedRows<'q>,
1608 b: KeyedRows<'q>,
1609 ra: Vec<Row>,
1610 rb: Vec<Row>,
1611 i: usize,
1612 j: usize,
1613}
1614
1615impl Iterator for MergeStream<'_> {
1616 type Item = Row;
1617
1618 fn next(&mut self) -> Option<Row> {
1619 loop {
1620 while self.i < self.ra.len() {
1621 if self.j < self.rb.len() {
1622 let j = self.j;
1623 self.j += 1;
1624 if let Some(m) = crate::row::merge_rows(&self.ra[self.i], &self.rb[j]) {
1625 return Some(m);
1626 }
1627 } else {
1628 self.i += 1;
1629 self.j = 0;
1630 }
1631 }
1632 let ka = self.a.peek()?.0;
1633 let kb = self.b.peek()?.0;
1634 match ka.cmp(&kb) {
1635 std::cmp::Ordering::Less => {
1636 self.a.next();
1637 }
1638 std::cmp::Ordering::Greater => {
1639 self.b.next();
1640 }
1641 std::cmp::Ordering::Equal => {
1642 self.ra.clear();
1643 self.rb.clear();
1644 while let Some((_, r)) = self.a.next_if(|(k, _)| *k == ka) {
1645 self.ra.push(r);
1646 }
1647 while let Some((_, r)) = self.b.next_if(|(k, _)| *k == ka) {
1648 self.rb.push(r);
1649 }
1650 self.i = 0;
1651 self.j = 0;
1652 }
1653 }
1654 }
1655 }
1656}
1657
1658fn probe_subject(ctx: &Ctx, t: &SlotTerm, base: &Row) -> Option<Option<u32>> {
1672 let dict = ctx.rete.dictionary();
1673 match t {
1674 SlotTerm::Node(n) => dict.node_as_subject_id(*n).map(Some),
1675 SlotTerm::Pred(_) => None,
1676 SlotTerm::Var(i) => match &base[*i] {
1677 None => Some(None),
1678 Some(Val::Id(v)) if *v >= 0 => dict.node_as_subject_id(*v as u32).map(Some),
1679 Some(_) => None,
1681 },
1682 }
1683}
1684
1685fn probe_object(ctx: &Ctx, t: &SlotTerm, base: &Row) -> Option<Option<u32>> {
1687 let dict = ctx.rete.dictionary();
1688 match t {
1689 SlotTerm::Node(n) => dict.node_as_object_id(*n).map(Some),
1690 SlotTerm::Pred(_) => None,
1691 SlotTerm::Var(i) => match &base[*i] {
1692 None => Some(None),
1693 Some(Val::Id(v)) if *v >= 0 => dict.node_as_object_id(*v as u32).map(Some),
1694 Some(_) => None,
1695 },
1696 }
1697}
1698
1699fn probe_predicate(ctx: &Ctx, t: &SlotTerm, base: &Row) -> Option<Option<u32>> {
1701 let dict = ctx.rete.dictionary();
1702 match t {
1703 SlotTerm::Pred(p) => Some(Some(*p)),
1704 SlotTerm::Node(_) => None,
1705 SlotTerm::Var(i) => match &base[*i] {
1706 None => Some(None),
1707 Some(Val::Id(v)) if *v < 0 => Some(Some((-v - 1) as u32)),
1708 Some(Val::Id(v)) => ctx
1710 .resolver
1711 .term(*v)
1712 .and_then(|t| dict.predicate_id(&t))
1713 .map(Some),
1714 Some(Val::Str(_)) => None,
1715 },
1716 }
1717}
1718
1719fn probe_rows<'q>(
1723 ctx: &'q Ctx<'q>,
1724 index: &'q GraphIndex,
1725 t: (SlotTerm, SlotTerm, SlotTerm),
1726 base: Row,
1727) -> Box<dyn Iterator<Item = Row> + 'q> {
1728 let (Some(sid), Some(pid), Some(oid)) = (
1729 probe_subject(ctx, &t.0, &base),
1730 probe_predicate(ctx, &t.1, &base),
1731 probe_object(ctx, &t.2, &base),
1732 ) else {
1733 return Box::new(std::iter::empty());
1734 };
1735 let dict = ctx.rete.dictionary();
1736 Box::new(
1737 index
1738 .scan_iter((sid, pid, oid))
1739 .filter_map(move |(s_id, p_id, o_id)| {
1740 let s_val = dict.subject_node(s_id) as i64;
1741 let p_val = pred_val(ctx, &t.1, p_id);
1742 let o_val = dict.object_node(o_id) as i64;
1743 let mut row = base.clone();
1744 for (term, val) in [(&t.0, s_val), (&t.1, p_val), (&t.2, o_val)] {
1745 if let SlotTerm::Var(i) = term {
1746 match row[*i] {
1747 Some(Val::Id(existing)) if existing != val => return None,
1748 Some(Val::Id(_)) => {}
1749 Some(Val::Str(_)) => return None,
1750 None => row[*i] = Some(Val::Id(val)),
1751 }
1752 }
1753 }
1754 Some(row)
1755 }),
1756 )
1757}
1758
1759pub(crate) struct ProbePlan {
1761 pats: Vec<(SlotTerm, SlotTerm, SlotTerm)>,
1762}
1763
1764impl ProbePlan {
1765 pub(crate) fn new(
1769 ctx: &Ctx,
1770 index: &GraphIndex,
1771 patterns: &[TriplePattern],
1772 seed_mask: &[bool],
1773 ) -> Option<Self> {
1774 let lowered = lower(patterns, ctx)?;
1775 let seed: std::collections::HashSet<usize> = seed_mask
1776 .iter()
1777 .enumerate()
1778 .filter_map(|(i, b)| b.then_some(i))
1779 .collect();
1780 let order = selectivity_order_seeded(ctx, index, &lowered, &seed);
1781 Some(ProbePlan {
1782 pats: order.into_iter().map(|i| lowered[i]).collect(),
1783 })
1784 }
1785}
1786
1787pub(crate) struct ProbeJoin<'q> {
1790 ctx: &'q Ctx<'q>,
1791 index: &'q GraphIndex,
1792 pats: Vec<(SlotTerm, SlotTerm, SlotTerm)>,
1793 stack: Vec<Box<dyn Iterator<Item = Row> + 'q>>,
1794}
1795
1796impl<'q> ProbeJoin<'q> {
1797 pub(crate) fn new(
1800 ctx: &'q Ctx<'q>,
1801 index: &'q GraphIndex,
1802 patterns: &[TriplePattern],
1803 ) -> Option<Self> {
1804 let plan = ProbePlan::new(ctx, index, patterns, &vec![false; ctx.slots.len()])?;
1805 Some(Self::from_plan(ctx, index, &plan, ctx.slots.empty_row()))
1806 }
1807
1808 pub(crate) fn from_plan(
1810 ctx: &'q Ctx<'q>,
1811 index: &'q GraphIndex,
1812 plan: &ProbePlan,
1813 seed: Row,
1814 ) -> Self {
1815 let pats = plan.pats.clone();
1816 let first = probe_rows(ctx, index, pats[0], seed);
1817 ProbeJoin {
1818 ctx,
1819 index,
1820 pats,
1821 stack: vec![first],
1822 }
1823 }
1824}
1825
1826impl Iterator for ProbeJoin<'_> {
1827 type Item = Row;
1828
1829 fn next(&mut self) -> Option<Row> {
1830 loop {
1831 let depth = self.stack.len();
1832 match self.stack.last_mut()?.next() {
1833 Some(row) => {
1834 if depth == self.pats.len() {
1835 return Some(row);
1836 }
1837 let it = probe_rows(self.ctx, self.index, self.pats[depth], row);
1838 self.stack.push(it);
1839 }
1840 None => {
1841 self.stack.pop();
1842 if self.stack.is_empty() {
1843 return None;
1844 }
1845 }
1846 }
1847 }
1848 }
1849}
1850
1851fn hash_join(
1855 left: Vec<Row>,
1856 left_bound: &[usize],
1857 right: Vec<Row>,
1858 right_slots: &[usize],
1859) -> Vec<Row> {
1860 if left.len() == 1 && left_bound.is_empty() {
1862 return right;
1863 }
1864 if left.is_empty() || right.is_empty() {
1865 return Vec::new();
1866 }
1867 let shared: Vec<usize> = right_slots
1868 .iter()
1869 .copied()
1870 .filter(|s| left_bound.contains(s))
1871 .collect();
1872 let fill = |l: &Row, r: &Row| -> Row {
1873 let mut out = l.clone();
1874 for &s in right_slots {
1875 out[s] = r[s].clone();
1876 }
1877 out
1878 };
1879 if shared.is_empty() {
1880 let mut out = Vec::with_capacity(left.len() * right.len());
1884 for l in &left {
1885 for r in &right {
1886 out.push(fill(l, r));
1887 }
1888 }
1889 return out;
1890 }
1891 let key_of = |b: &Row| -> Vec<Val> { shared.iter().map(|&s| b[s].clone().unwrap()).collect() };
1892 let mut out = Vec::new();
1899 if right.len() <= left.len() {
1900 let mut buckets: HashMap<Vec<Val>, Vec<Row>> = HashMap::new();
1901 for r in right {
1902 buckets.entry(key_of(&r)).or_default().push(r);
1903 }
1904 for l in &left {
1905 if let Some(rs) = buckets.get(&key_of(l)) {
1906 for r in rs {
1907 out.push(fill(l, r));
1908 }
1909 }
1910 }
1911 } else {
1912 let mut buckets: HashMap<Vec<Val>, Vec<Row>> = HashMap::new();
1914 for l in left {
1915 buckets.entry(key_of(&l)).or_default().push(l);
1916 }
1917 for r in &right {
1918 if let Some(ls) = buckets.get(&key_of(r)) {
1919 for l in ls {
1920 let mut row = r.clone();
1921 for &s in left_bound {
1922 if let Some(v) = &l[s] {
1923 row[s] = Some(v.clone());
1924 }
1925 }
1926 out.push(row);
1927 }
1928 }
1929 }
1930 }
1931 out
1932}
1933
1934fn const_subject(t: &SlotTerm, d: &crate::Dictionary) -> Option<Option<u32>> {
1939 match t {
1940 SlotTerm::Node(n) => d.node_as_subject_id(*n).map(Some),
1941 SlotTerm::Pred(_) => None, SlotTerm::Var(_) => Some(None),
1943 }
1944}
1945
1946fn const_object(t: &SlotTerm, d: &crate::Dictionary) -> Option<Option<u32>> {
1947 match t {
1948 SlotTerm::Node(n) => d.node_as_object_id(*n).map(Some),
1949 SlotTerm::Pred(_) => None,
1950 SlotTerm::Var(_) => Some(None),
1951 }
1952}
1953
1954fn const_predicate(t: &SlotTerm) -> Option<Option<u32>> {
1955 match t {
1956 SlotTerm::Pred(p) => Some(Some(*p)),
1957 SlotTerm::Node(_) => None, SlotTerm::Var(_) => Some(None),
1959 }
1960}
1961
1962#[cfg(test)]
1963mod tests {
1964 use super::*;
1965 use crate::dictionary::DictionaryBuilder;
1966 use crate::file::write_file_v05 as write_file;
1967 use crate::index::GraphIndexBuilder;
1968
1969 fn rete_from(triples: &[(&str, &str, &str)]) -> Vec<u8> {
1970 let mut db = DictionaryBuilder::new();
1971 for (s, p, o) in triples {
1972 db.observe(s, p, o);
1973 }
1974 let dict = db.build();
1975 let mut ib = GraphIndexBuilder::new();
1976 for (s, p, o) in triples {
1977 ib.push(dict.encode(s, p, o).unwrap());
1978 }
1979 write_file(&dict, &ib.build(), false, &[], 0)
1980 }
1981
1982 fn pat(s: &str, p: &str, o: &str) -> TriplePattern {
1983 TriplePattern {
1984 s: PatternTerm::parse(s),
1985 p: PatternTerm::parse(p),
1986 o: PatternTerm::parse(o),
1987 }
1988 }
1989
1990 #[test]
1999 fn scan_bytes_route_a_literal_by_its_object_id() {
2000 let value = "<http://ex/value>";
2001 let owned: Vec<(String, String)> = (0..400)
2002 .map(|i| (format!("<http://ex/s{i:03}>"), format!("\"lit {i:03}\"")))
2003 .collect();
2004 let mut db = DictionaryBuilder::new();
2005 for (s, o) in &owned {
2006 db.observe(s, value, o);
2007 }
2008 let dict = db.build();
2009 let mut ib = GraphIndexBuilder::new().with_tile_budget(256);
2010 for (s, o) in &owned {
2011 ib.push(dict.encode(s, value, o).unwrap());
2012 }
2013 let rete = Rete::open(&write_file(&dict, &ib.build(), false, &[], 0)).unwrap();
2014 let index = rete.default_index();
2015 let whole = routed_scan_bytes(index, (None, rete.dictionary().predicate_id(value), None));
2017 for (_, lit) in owned.iter().step_by(57) {
2018 let q = [pat("?s", value, lit)];
2019 let mut slots = Slots::new();
2020 collect_pattern_slots(&q, &mut slots);
2021 let ctx = Ctx::new(&rete, slots);
2022 let got = bgp_min_scan_bytes(&ctx, index, &q).unwrap();
2023 assert!(got > 0, "{lit} routed to no tile");
2024 assert!(
2025 got < whole,
2026 "{lit} routed to the whole predicate ({got} of {whole})"
2027 );
2028 }
2029 }
2030
2031 #[test]
2037 fn plan_gates_do_not_move_with_the_compression() {
2038 let mut owned: Vec<(String, String, String)> = Vec::new();
2039 for i in 0..3000 {
2040 let s = format!("<http://ex/s{i}>");
2041 owned.push((
2042 s.clone(),
2043 "<http://ex/kind>".into(),
2044 format!("<http://ex/K{}>", i % 7),
2045 ));
2046 owned.push((s.clone(), "<http://ex/value>".into(), format!("\"v {i}\"")));
2047 if i % 50 == 0 {
2048 owned.push((s, "<http://ex/rare>".into(), "<http://ex/R>".into()));
2049 }
2050 }
2051 let mut db = DictionaryBuilder::new();
2052 for (s, p, o) in &owned {
2053 db.observe(s, p, o);
2054 }
2055 let dict = db.build();
2056 let mut ib = GraphIndexBuilder::new().with_tile_budget(1024);
2057 for (s, p, o) in &owned {
2058 ib.push(dict.encode(s, p, o).unwrap());
2059 }
2060 let bytes = write_file(&dict, &ib.build(), false, &[], 0);
2061 let leaked: &'static [u8] = Box::leak(bytes.clone().into_boxed_slice());
2062 let resident = Rete::open(&bytes).unwrap();
2063 let lazy = Rete::open_ranged_lazy(crate::reader::SliceReader::new(leaked)).unwrap();
2064 let shapes: Vec<Vec<TriplePattern>> = vec![
2065 vec![
2066 pat("?s", "<http://ex/rare>", "<http://ex/R>"),
2067 pat("?s", "<http://ex/value>", "?v"),
2068 ],
2069 vec![
2070 pat("?s", "<http://ex/kind>", "<http://ex/K3>"),
2071 pat("?s", "<http://ex/value>", "?v"),
2072 ],
2073 vec![
2074 pat("?s", "<http://ex/kind>", "?k"),
2075 pat("?s", "<http://ex/value>", "?v"),
2076 ],
2077 vec![
2078 pat("?s", "<http://ex/value>", "\"v 7\""),
2079 pat("?s", "<http://ex/kind>", "?k"),
2080 ],
2081 ];
2082 let decisions = |rete: &Rete| -> Vec<(Vec<u64>, bool, Option<u64>)> {
2083 shapes
2084 .iter()
2085 .map(|q| {
2086 let mut slots = Slots::new();
2087 collect_pattern_slots(q, &mut slots);
2088 let ctx = Ctx::new(rete, slots);
2089 let index = rete.default_index();
2090 let lowered = lower(q, &ctx).unwrap();
2091 let est = tile_estimates(&ctx, index, &lowered, &Default::default());
2092 let lopsided = merge_seed_lopsided(&ctx, index, &lowered[0], &lowered[1]);
2093 let min = bgp_min_estimate(&ctx, index, q).map(|e| e as u64);
2094 (est.iter().map(|e| *e as u64).collect(), lopsided, min)
2095 })
2096 .collect()
2097 };
2098 assert_eq!(decisions(&resident), decisions(&lazy));
2099 #[cfg(feature = "compression")]
2101 {
2102 let bytes_of = |rete: &Rete| {
2103 let q = &shapes[2];
2104 let mut slots = Slots::new();
2105 collect_pattern_slots(q, &mut slots);
2106 let ctx = Ctx::new(rete, slots);
2107 bgp_min_scan_bytes(&ctx, rete.default_index(), q)
2108 };
2109 assert_ne!(bytes_of(&resident), bytes_of(&lazy));
2110 }
2111 }
2112
2113 #[test]
2123 fn a_fat_predicate_estimates_alike_in_both_layouts() {
2124 let mut owned: Vec<(String, String, String)> = Vec::new();
2125 for i in 0..120_000 {
2126 let s = format!("<http://ex/s{i:06}>");
2127 owned.push((s.clone(), "<http://ex/kind>".into(), "<http://ex/K>".into()));
2128 owned.push((s, "<http://ex/value>".into(), format!("\"v {i}\"")));
2129 }
2130 let mut db = DictionaryBuilder::new();
2131 for (s, p, o) in &owned {
2132 db.observe(s, p, o);
2133 }
2134 let dict = db.build();
2135 let mut ib = GraphIndexBuilder::new().with_tile_budget(256);
2136 for (s, p, o) in &owned {
2137 ib.push(dict.encode(s, p, o).unwrap());
2138 }
2139 let v5 = write_file(&dict, &ib.build(), false, &[], 0);
2140 let v6: &'static [u8] = Box::leak(
2141 crate::v06::write::convert_image(&v5, Default::default())
2142 .unwrap()
2143 .into_boxed_slice(),
2144 );
2145 let leaked5: &'static [u8] = Box::leak(v5.clone().into_boxed_slice());
2146 let flat = Rete::open_ranged_lazy(crate::reader::SliceReader::new(leaked5)).unwrap();
2147 let leafed = Rete::open_ranged_lazy(crate::reader::SliceReader::new(v6)).unwrap();
2148 let q = [pat("?s", "<http://ex/value>", "?v")];
2149 let figures = |rete: &Rete| {
2150 let mut slots = Slots::new();
2151 collect_pattern_slots(&q, &mut slots);
2152 let ctx = Ctx::new(rete, slots);
2153 let index = rete.default_index();
2154 let total = bgp_total_estimate(&ctx, index, &q).unwrap();
2155 (total, probe_row_budget(total))
2156 };
2157 let (est5, budget5) = figures(&flat);
2158 let (est6, budget6) = figures(&leafed);
2159 assert!(est5 >= 70_000.0, "flat estimate {est5}");
2160 assert!(
2161 est6 >= est5 / 2.0 && est6 <= est5 * 2.0,
2162 "0x06 estimates {est6} where 0x05 estimates {est5}"
2163 );
2164 assert!(budget5.is_some_and(|b| b >= 1000), "{budget5:?}");
2166 assert!(budget6.is_some_and(|b| b >= 1000), "{budget6:?}");
2167 }
2168
2169 #[test]
2176 fn a_rare_object_of_a_fat_predicate_estimates_small_in_both_layouts() {
2177 let ty = "<http://www.w3.org/1999/02/22-rdf-syntax-ns#type>";
2178 let mut owned: Vec<(String, String, String)> = Vec::new();
2179 for i in 0..120_000 {
2180 let s = format!("<http://ex/s{i:06}>");
2181 owned.push((s.clone(), ty.into(), "<http://ex/Common>".into()));
2182 owned.push((s.clone(), "<http://ex/other>".into(), format!("\"o {i}\"")));
2183 if i % 3000 == 0 {
2184 owned.push((s, ty.into(), "<http://ex/Rare>".into()));
2185 }
2186 }
2187 let mut db = DictionaryBuilder::new();
2188 for (s, p, o) in &owned {
2189 db.observe(s, p, o);
2190 }
2191 let dict = db.build();
2192 let mut ib = GraphIndexBuilder::new().with_tile_budget(256);
2193 for (s, p, o) in &owned {
2194 ib.push(dict.encode(s, p, o).unwrap());
2195 }
2196 let v5 = write_file(&dict, &ib.build(), false, &[], 0);
2197 let v6: &'static [u8] = Box::leak(
2198 crate::v06::write::convert_image(&v5, Default::default())
2199 .unwrap()
2200 .into_boxed_slice(),
2201 );
2202 let leaked5: &'static [u8] = Box::leak(v5.into_boxed_slice());
2203 let figures = |rete: &Rete| {
2204 let est = |q: &[TriplePattern]| {
2205 let mut slots = Slots::new();
2206 collect_pattern_slots(q, &mut slots);
2207 let ctx = Ctx::new(rete, slots);
2208 bgp_min_estimate(&ctx, rete.default_index(), q).unwrap()
2209 };
2210 (
2211 est(&[pat("?s", ty, "<http://ex/Rare>")]),
2212 est(&[pat("?s", ty, "<http://ex/Common>")]),
2213 est(&[pat("?s", "<http://ex/other>", "?x")]),
2214 )
2215 };
2216 for (name, bytes) in [("0x05", leaked5), ("0x06", v6)] {
2217 let rete = Rete::open_ranged_lazy(crate::reader::SliceReader::new(bytes)).unwrap();
2218 let (rare, common, other) = figures(&rete);
2219 assert!(rare * 20.0 < other, "{name}: rare {rare} vs other {other}");
2220 assert!(
2221 rare * 20.0 < common,
2222 "{name}: rare {rare} vs common {common}"
2223 );
2224 }
2225 }
2226
2227 #[test]
2232 fn a_tile_span_is_the_same_in_both_layouts() {
2233 let ty = "<http://www.w3.org/1999/02/22-rdf-syntax-ns#type>";
2234 let mut owned: Vec<(String, String, String)> = Vec::new();
2235 for i in 0..30_000 {
2236 let s = format!("<http://ex/s{i:06}>");
2237 owned.push((s.clone(), ty.into(), format!("<http://ex/C{}>", i % 7)));
2238 owned.push((s, "<http://ex/other>".into(), format!("\"o {}\"", i % 5000)));
2239 }
2240 let mut db = DictionaryBuilder::new();
2241 for (s, p, o) in &owned {
2242 db.observe(s, p, o);
2243 }
2244 let dict = db.build();
2245 let mut ib = GraphIndexBuilder::new().with_tile_budget(256);
2246 for (s, p, o) in &owned {
2247 ib.push(dict.encode(s, p, o).unwrap());
2248 }
2249 let v5 = write_file(&dict, &ib.build(), false, &[], 0);
2250 let v6 = crate::v06::write::convert_image(&v5, Default::default()).unwrap();
2251 let (v5, v6): (&'static [u8], &'static [u8]) = (
2252 Box::leak(v5.into_boxed_slice()),
2253 Box::leak(v6.into_boxed_slice()),
2254 );
2255 let flat = Rete::open_ranged_lazy(crate::reader::SliceReader::new(v5)).unwrap();
2256 let leafed = Rete::open_ranged_lazy(crate::reader::SliceReader::new(v6)).unwrap();
2257 let (fi, li) = (flat.default_index(), leafed.default_index());
2258 let mut checked = 0;
2259 for perm in fi.perms().iter() {
2260 let si = perm.section_index();
2261 for a in (0..dict.term_count() + 2).step_by(97) {
2262 assert_eq!(
2263 fi.tile_span(si, Some(a)),
2264 li.tile_span(si, Some(a)),
2265 "section {si}, id {a}"
2266 );
2267 checked += 1;
2268 }
2269 assert_eq!(fi.tile_span(si, None), li.tile_span(si, None));
2270 }
2271 assert!(checked > 100);
2272 }
2273
2274 #[test]
2279 fn every_join_path_returns_the_same_solutions() {
2280 let mut owned: Vec<(String, String, String)> = Vec::new();
2281 let mut x: u64 = 7;
2282 let mut next = |m: u64| {
2283 x = x
2284 .wrapping_mul(6364136223846793005)
2285 .wrapping_add(1442695040888963407);
2286 (x >> 33) % m
2287 };
2288 for i in 0..60 {
2289 let s = format!("s{i}");
2290 owned.push((s.clone(), "a".into(), format!("C{}", next(3))));
2291 for p in 0..4 {
2292 for _ in 0..next(3) {
2293 owned.push((s.clone(), format!("p{p}"), format!("s{}", next(60))));
2294 }
2295 }
2296 }
2297 let triples: Vec<(&str, &str, &str)> = owned
2298 .iter()
2299 .map(|(s, p, o)| (s.as_str(), p.as_str(), o.as_str()))
2300 .collect();
2301 let bytes = rete_from(&triples);
2302 let rete = Rete::open(&bytes).unwrap();
2303 let queries: Vec<Vec<TriplePattern>> = vec![
2304 vec![pat("?s", "p0", "?a"), pat("?s", "p1", "?b")],
2305 vec![pat("?s", "a", "C1"), pat("?s", "p0", "?a")],
2306 vec![pat("?s", "p0", "?a"), pat("?a", "p2", "?b")],
2307 vec![pat("s5", "p0", "?a"), pat("?a", "?p", "?o")],
2308 vec![pat("?s", "p0", "?x"), pat("?s", "p1", "?x")],
2309 vec![
2310 pat("?s", "p3", "?o"),
2311 pat("?o", "a", "C0"),
2312 pat("?s", "p1", "?z"),
2313 ],
2314 vec![
2315 pat("?s", "p0", "?a"),
2316 pat("?s", "p1", "?b"),
2317 pat("?b", "p2", "?c"),
2318 ],
2319 ];
2320 let run = |limits: Option<PlanLimits>| -> Vec<Vec<Vec<(String, String)>>> {
2321 TEST_LIMITS.with(|c| c.set(limits));
2322 let out = queries
2323 .iter()
2324 .map(|q| {
2325 let mut slots = crate::row::Slots::new();
2326 collect_pattern_slots(q, &mut slots);
2327 let ctx = Ctx::new(&rete, slots);
2328 let mut rows: Vec<Vec<(String, String)>> =
2329 BgpSolutions::new(&ctx, rete.default_index(), q)
2330 .map(|r| row_to_binding(&ctx, &r).into_iter().collect())
2331 .collect();
2332 rows.sort();
2333 rows
2334 })
2335 .collect();
2336 TEST_LIMITS.with(|c| c.set(None));
2337 out
2338 };
2339 let reference = run(None);
2340 assert!(reference.iter().all(|r| !r.is_empty()), "{reference:?}");
2341 for (name, l) in [
2342 (
2343 "probe everything",
2344 PlanLimits {
2345 probe_floor: 0.0,
2346 stream_seed: f64::MAX,
2347 hash_build: f64::MAX,
2348 },
2349 ),
2350 (
2351 "stream + hash",
2352 PlanLimits {
2353 probe_floor: f64::MAX,
2354 stream_seed: 0.0,
2355 hash_build: f64::MAX,
2356 },
2357 ),
2358 (
2359 "stream + probe",
2360 PlanLimits {
2361 probe_floor: f64::MAX,
2362 stream_seed: 0.0,
2363 hash_build: 0.0,
2364 },
2365 ),
2366 (
2367 "stream, all paths",
2368 PlanLimits {
2369 probe_floor: 0.0,
2370 stream_seed: 0.0,
2371 hash_build: 30.0,
2372 },
2373 ),
2374 ] {
2375 assert_eq!(run(Some(l)), reference, "{name}");
2376 }
2377 }
2378
2379 #[test]
2380 fn single_pattern_binds_variable() {
2381 let bytes = rete_from(&[("Alice", "knows", "Bob"), ("Bob", "knows", "Carol")]);
2382 let rete = Rete::open(&bytes).unwrap();
2383 let sols = eval_bgp(&rete, &[pat("Alice", "knows", "?y")]);
2384 assert_eq!(sols.len(), 1);
2385 assert_eq!(sols[0]["y"], "Bob");
2386 }
2387
2388 #[test]
2389 fn two_hop_join_on_shared_variable() {
2390 let bytes = rete_from(&[
2392 ("Alice", "knows", "Bob"),
2393 ("Bob", "knows", "Carol"),
2394 ("Carol", "knows", "Dave"),
2395 ("Alice", "knows", "Eve"), ]);
2397 let rete = Rete::open(&bytes).unwrap();
2398 let sols = eval_bgp(&rete, &[pat("?x", "knows", "?y"), pat("?y", "knows", "?z")]);
2399 let mut got: Vec<_> = sols
2401 .iter()
2402 .map(|b| (b["x"].clone(), b["y"].clone(), b["z"].clone()))
2403 .collect();
2404 got.sort();
2405 assert_eq!(
2406 got,
2407 vec![
2408 ("Alice".into(), "Bob".into(), "Carol".into()),
2409 ("Bob".into(), "Carol".into(), "Dave".into()),
2410 ]
2411 );
2412 }
2413
2414 #[test]
2415 fn repeated_variable_within_pattern() {
2416 let bytes = rete_from(&[("Alice", "knows", "Bob"), ("Bob", "knows", "Bob")]);
2418 let rete = Rete::open(&bytes).unwrap();
2419 let sols = eval_bgp(&rete, &[pat("?x", "knows", "?x")]);
2420 assert_eq!(sols.len(), 1);
2421 assert_eq!(sols[0]["x"], "Bob");
2422 }
2423
2424 #[test]
2425 fn no_solutions_yields_empty() {
2426 let bytes = rete_from(&[("Alice", "knows", "Bob")]);
2427 let rete = Rete::open(&bytes).unwrap();
2428 assert!(eval_bgp(&rete, &[pat("Alice", "likes", "?y")]).is_empty());
2429 }
2430
2431 #[test]
2438 fn snowflake_type_and_adjacency_matches_reference() {
2439 const TYPE: &str = "<http://www.w3.org/1999/02/22-rdf-syntax-ns#type>";
2440 let mut triples: Vec<(String, String, String)> = Vec::new();
2441 for i in 0..60 {
2443 let s = format!("S{i}");
2444 triples.push((s.clone(), TYPE.into(), "Student".into()));
2445 triples.push((s.clone(), "takesCourse".into(), format!("c{}", i % 5)));
2446 triples.push((s, "takesCourse".into(), format!("c{}", (i + 1) % 5)));
2447 }
2448 for j in 0..5 {
2450 triples.push((format!("c{j}"), TYPE.into(), "Course".into()));
2451 }
2452 triples.push(("Prof".into(), "teacherOf".into(), "c0".into()));
2453 triples.push(("Prof".into(), "teacherOf".into(), "c2".into()));
2454
2455 let refs: Vec<(&str, &str, &str)> = triples
2456 .iter()
2457 .map(|(s, p, o)| (s.as_str(), p.as_str(), o.as_str()))
2458 .collect();
2459 let bytes = rete_from(&refs);
2460 let rete = Rete::open(&bytes).unwrap();
2461
2462 let got: std::collections::BTreeSet<(String, String)> = eval_bgp(
2463 &rete,
2464 &[
2465 pat("?x", TYPE, "Student"),
2466 pat("?y", TYPE, "Course"),
2467 pat("?x", "takesCourse", "?y"),
2468 pat("Prof", "teacherOf", "?y"),
2469 ],
2470 )
2471 .into_iter()
2472 .map(|b| (b["x"].clone(), b["y"].clone()))
2473 .collect();
2474
2475 let taught = ["c0", "c2"];
2477 let mut want = std::collections::BTreeSet::new();
2478 for i in 0..60 {
2479 for c in [i % 5, (i + 1) % 5] {
2480 let course = format!("c{c}");
2481 if taught.contains(&course.as_str()) {
2482 want.insert((format!("S{i}"), course));
2483 }
2484 }
2485 }
2486 assert_eq!(got, want, "snowflake join must match the brute-force set");
2487 assert!(!want.is_empty(), "sanity: the reference set is non-empty");
2488 }
2489
2490 #[test]
2495 fn merge_join_subject_star_cross_product() {
2496 let bytes = rete_from(&[
2497 ("x", "a", "a1"),
2498 ("x", "a", "a2"),
2499 ("x", "b", "b1"),
2500 ("x", "b", "b2"),
2501 ("y", "a", "a3"),
2502 ("y", "b", "b3"),
2503 ("z", "a", "a4"), ]);
2505 let rete = Rete::open(&bytes).unwrap();
2506 let mut got: Vec<(String, String, String)> =
2507 eval_bgp(&rete, &[pat("?x", "a", "?av"), pat("?x", "b", "?bv")])
2508 .iter()
2509 .map(|m| (m["x"].clone(), m["av"].clone(), m["bv"].clone()))
2510 .collect();
2511 got.sort();
2512 let mut want: Vec<(String, String, String)> = [
2513 ("x", "a1", "b1"),
2514 ("x", "a1", "b2"),
2515 ("x", "a2", "b1"),
2516 ("x", "a2", "b2"),
2517 ("y", "a3", "b3"),
2518 ]
2519 .into_iter()
2520 .map(|(a, b, c)| (a.to_string(), b.to_string(), c.to_string()))
2521 .collect();
2522 want.sort();
2523 assert_eq!(got, want, "same-role merge must equal the brute-force join");
2524 }
2525}