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 let dict = ctx.rete.dictionary();
795 let s_val = dict.subject_node(s_id) as i64;
796 let p_val = pred_val(ctx, &t.1, p_id);
797 let o_val = dict.object_node(o_id) as i64;
798 let mut row = ctx.slots.empty_row();
799 for (term, val) in [(&t.0, s_val), (&t.1, p_val), (&t.2, o_val)] {
800 if let SlotTerm::Var(i) = term {
801 match row[*i] {
802 Some(Val::Id(existing)) if existing != val => return None,
803 Some(_) => {}
804 None => row[*i] = Some(Val::Id(val)),
805 }
806 }
807 }
808 Some(row)
809}
810
811fn scan_rows<'q>(
816 ctx: &'q Ctx<'q>,
817 index: &'q GraphIndex,
818 t: (SlotTerm, SlotTerm, SlotTerm),
819) -> Option<impl Iterator<Item = Row> + 'q> {
820 let dict = ctx.rete.dictionary();
821 let (sid, pid, oid) = (
822 const_subject(&t.0, dict)?,
823 const_predicate(&t.1)?,
824 const_object(&t.2, dict)?,
825 );
826 Some(
827 index
828 .scan_iter((sid, pid, oid))
829 .filter_map(move |triple| triple_row(ctx, &t, triple)),
830 )
831}
832
833fn pattern_rows(
836 ctx: &Ctx,
837 index: &GraphIndex,
838 t: &(SlotTerm, SlotTerm, SlotTerm),
839) -> Option<(Vec<Row>, Vec<usize>)> {
840 let slots = pattern_slots(t);
841 let mut rel: Vec<Row> = Vec::new();
842 let dict = ctx.rete.dictionary();
843 let (sid, pid, oid) = (
844 const_subject(&t.0, dict)?,
845 const_predicate(&t.1)?,
846 const_object(&t.2, dict)?,
847 );
848 for triple in index.scan_iter((sid, pid, oid)) {
851 if let Some(row) = triple_row(ctx, t, triple) {
852 rel.push(row);
853 }
854 }
855 Some((rel, slots))
856}
857
858fn pattern_estimates(
870 ctx: &Ctx,
871 index: &GraphIndex,
872 lowered: &[(SlotTerm, SlotTerm, SlotTerm)],
873 seed: &std::collections::HashSet<usize>,
874) -> Option<Vec<f64>> {
875 let pyr = ctx.rete.pyramid_if_loaded()?;
878 let mut pred: std::collections::HashMap<u32, u64> = std::collections::HashMap::new();
879 for e in &pyr.summary {
880 *pred.entry(e.predicate).or_insert(0) += e.count as u64;
881 }
882 let stats: std::collections::HashMap<u32, &crate::meta::PredStat> = ctx
889 .rete
890 .planner_stats(true)
891 .iter()
892 .map(|s| (s.predicate, s))
893 .collect();
894 let graph_total = ctx.rete.index_triples(index).max(1) as f64;
897 let total = graph_total;
898 let num_preds = pred.len().max(1) as f64;
899 const SEL_SUBJECT: f64 = 0.001;
901 const SEL_OBJECT: f64 = 0.02;
902 let node_bound = |t: &SlotTerm| match t {
903 SlotTerm::Node(_) => true,
904 SlotTerm::Var(v) => seed.contains(v),
905 SlotTerm::Pred(_) => false,
906 };
907 Some(
908 lowered
909 .iter()
910 .map(|t| {
911 let (base, st) = match t.1 {
916 SlotTerm::Pred(p) => {
917 (*pred.get(&p).unwrap_or(&0) as f64, stats.get(&p).copied())
918 }
919 SlotTerm::Var(v) if seed.contains(&v) => (total / num_preds, None),
920 _ => (total, None),
921 };
922 let mut est = base.min(graph_total).max(1.0);
923 if node_bound(&t.0) {
924 est *= match st {
925 Some(s) if s.distinct_subjects > 0 => 1.0 / s.distinct_subjects as f64,
926 _ => SEL_SUBJECT,
927 };
928 }
929 if node_bound(&t.2) {
930 est *= match st {
931 Some(s) if s.distinct_objects > 0 => 1.0 / s.distinct_objects as f64,
932 _ => SEL_OBJECT,
933 };
934 }
935 est.max(1.0)
936 })
937 .collect(),
938 )
939}
940
941pub(crate) fn bgp_min_scan_bytes(
948 ctx: &Ctx,
949 index: &GraphIndex,
950 patterns: &[TriplePattern],
951) -> Option<u64> {
952 let lowered = lower(patterns, ctx)?;
953 lowered
954 .iter()
955 .map(|t| pattern_scan_bytes(ctx, index, t))
956 .min()
957}
958
959fn pattern_scan_bytes(ctx: &Ctx, index: &GraphIndex, t: &(SlotTerm, SlotTerm, SlotTerm)) -> u64 {
976 let dict = ctx.rete.dictionary();
977 match (
978 const_subject(&t.0, dict),
979 const_predicate(&t.1),
980 const_object(&t.2, dict),
981 ) {
982 (Some(s), Some(p), Some(o)) => routed_scan_bytes(index, (s, p, o)),
983 _ => 0,
984 }
985}
986
987pub(crate) fn routed_scan_bytes(index: &GraphIndex, pattern: Pattern) -> u64 {
989 let sections = index.tile_sections();
990 let mut best = u64::MAX;
991 for perm in index.perms().iter() {
995 let tiles = sections[perm.section_index()];
996 best = best.min(match perm.order_pattern(pattern)[0] {
999 Some(id) => tiles.routed_bytes(id),
1000 None => tiles.total_bytes(),
1001 });
1002 }
1003 best
1004}
1005
1006pub(crate) const FAT_SCAN_ROWS: f64 = (1u64 << 20) as f64;
1019
1020fn merge_seed_lopsided(
1025 ctx: &Ctx,
1026 index: &GraphIndex,
1027 ta: &(SlotTerm, SlotTerm, SlotTerm),
1028 tb: &(SlotTerm, SlotTerm, SlotTerm),
1029) -> bool {
1030 let est = tile_estimates(ctx, index, &[*ta, *tb], &std::collections::HashSet::new());
1031 let (lo, hi) = (est[0].min(est[1]), est[0].max(est[1]));
1032 hi >= FAT_SCAN_ROWS && hi / 4.0 >= lo
1033}
1034
1035fn selectivity_order(
1042 ctx: &Ctx,
1043 index: &GraphIndex,
1044 lowered: &[(SlotTerm, SlotTerm, SlotTerm)],
1045) -> Vec<usize> {
1046 selectivity_order_seeded(ctx, index, lowered, &std::collections::HashSet::new())
1047}
1048
1049fn selectivity_order_seeded(
1053 ctx: &Ctx,
1054 index: &GraphIndex,
1055 lowered: &[(SlotTerm, SlotTerm, SlotTerm)],
1056 seed: &std::collections::HashSet<usize>,
1057) -> Vec<usize> {
1058 let (estimates, per_class) = match pattern_estimates(ctx, index, lowered, seed) {
1066 Some(e) => (Some(e), false),
1067 None => (Some(tile_estimates(ctx, index, lowered, seed)), true),
1068 };
1069 let consts = |t: &(SlotTerm, SlotTerm, SlotTerm)| {
1070 [&t.0, &t.1, &t.2]
1071 .into_iter()
1072 .filter(|x| match x {
1073 SlotTerm::Var(v) => seed.contains(v),
1074 _ => true,
1075 })
1076 .count()
1077 };
1078 let vars = |t: &(SlotTerm, SlotTerm, SlotTerm)| -> Vec<usize> {
1079 [&t.0, &t.1, &t.2]
1080 .into_iter()
1081 .filter_map(|x| match x {
1082 SlotTerm::Var(v) => Some(*v),
1083 _ => None,
1084 })
1085 .collect()
1086 };
1087 let score = |i: usize| -> f64 {
1091 match &estimates {
1092 Some(e) => -e[i],
1093 None => consts(&lowered[i]) as f64,
1094 }
1095 };
1096 let type_pid = ctx.rete.dictionary().predicate_id(crate::file::RDF_TYPE);
1104 let is_class_enum = |i: usize, bound: &std::collections::HashSet<usize>| -> bool {
1105 let t = &lowered[i];
1106 !per_class
1107 && type_pid.is_some_and(|tp| matches!(t.1, SlotTerm::Pred(p) if p == tp))
1108 && matches!(t.2, SlotTerm::Node(_))
1109 && matches!(t.0, SlotTerm::Var(v) if !bound.contains(&v))
1110 };
1111 let n = lowered.len();
1112 let mut remaining: Vec<usize> = (0..n).collect();
1113 let mut order: Vec<usize> = Vec::with_capacity(n);
1114 let mut bound: std::collections::HashSet<usize> = seed.clone();
1115 while !remaining.is_empty() {
1116 let best = *remaining
1119 .iter()
1120 .max_by(|&&a, &&b| {
1121 let connected = |i: usize| vars(&lowered[i]).iter().any(|v| bound.contains(v));
1122 connected(a)
1123 .cmp(&connected(b))
1124 .then_with(|| is_class_enum(b, &bound).cmp(&is_class_enum(a, &bound)))
1125 .then_with(|| {
1126 score(a)
1127 .partial_cmp(&score(b))
1128 .unwrap_or(std::cmp::Ordering::Equal)
1129 })
1130 .then_with(|| b.cmp(&a))
1131 })
1132 .unwrap();
1133 for v in vars(&lowered[best]) {
1134 bound.insert(v);
1135 }
1136 order.push(best);
1137 remaining.retain(|&i| i != best);
1138 }
1139 order
1140}
1141
1142pub(crate) fn bgp_exists(ctx: &Ctx, index: &GraphIndex, patterns: &[TriplePattern]) -> bool {
1148 let dict = ctx.rete.dictionary();
1149 let Some(lowered) = lower(patterns, ctx) else {
1150 return false;
1151 };
1152 if let [t] = lowered.as_slice() {
1153 let names: Vec<usize> = [&t.0, &t.1, &t.2]
1157 .into_iter()
1158 .filter_map(|x| match x {
1159 SlotTerm::Var(v) => Some(*v),
1160 _ => None,
1161 })
1162 .collect();
1163 let distinct = names
1164 .iter()
1165 .enumerate()
1166 .all(|(i, v)| !names[i + 1..].contains(v));
1167 if distinct {
1168 return match (
1169 const_subject(&t.0, dict),
1170 const_predicate(&t.1),
1171 const_object(&t.2, dict),
1172 ) {
1173 (Some(s), Some(p), Some(o)) => index.scan_iter((s, p, o)).next().is_some(),
1174 _ => false,
1175 };
1176 }
1177 }
1178 !eval_bgp_rows(ctx, index, patterns).is_empty()
1179}
1180
1181pub(crate) struct BgpSolutions<'q> {
1192 scan: Option<Box<dyn Iterator<Item = Row> + 'q>>,
1193 prefix: Vec<Row>,
1194 buckets: HashMap<Vec<Val>, Vec<usize>>,
1196 shared: Vec<usize>,
1197 cartesian: bool,
1198 seed_only: bool,
1201 cur_scan: Option<Row>,
1202 matches: Vec<usize>,
1204 mi: usize,
1205}
1206
1207impl<'q> BgpSolutions<'q> {
1208 fn passthrough(rows: Box<dyn Iterator<Item = Row> + 'q>) -> Self {
1210 BgpSolutions {
1211 scan: Some(rows),
1212 prefix: Vec::new(),
1213 buckets: HashMap::new(),
1214 shared: Vec::new(),
1215 cartesian: false,
1216 seed_only: true,
1217 cur_scan: None,
1218 matches: Vec::new(),
1219 mi: 0,
1220 }
1221 }
1222
1223 fn streaming(
1232 ctx: &'q Ctx<'q>,
1233 index: &'q GraphIndex,
1234 lowered: &[(SlotTerm, SlotTerm, SlotTerm)],
1235 order: &[usize],
1236 est: &[f64],
1237 ) -> Self {
1238 let (mut stream, mut bound, skip): (Box<dyn Iterator<Item = Row> + 'q>, Vec<usize>, usize) =
1239 match (est[order[1]] > limits().stream_seed)
1240 .then(|| merge_stream(ctx, index, lowered[order[0]], lowered[order[1]]))
1241 .flatten()
1242 {
1243 Some((m, slots)) => (m, slots, 2),
1244 None => {
1245 let t = lowered[order[0]];
1246 let Some(scan) = scan_rows(ctx, index, t) else {
1247 return Self::empty();
1248 };
1249 (Box::new(scan), pattern_slots(&t), 1)
1250 }
1251 };
1252 for &i in &order[skip..] {
1253 let t = lowered[i];
1254 let slots = pattern_slots(&t);
1255 let shared: Vec<usize> = slots
1256 .iter()
1257 .copied()
1258 .filter(|s| bound.contains(s))
1259 .collect();
1260 if !shared.is_empty() && est[i] > limits().hash_build {
1261 stream = Box::new(stream.flat_map(move |base| probe_rows(ctx, index, t, base)));
1262 } else {
1263 let Some((rel, _)) = pattern_rows(ctx, index, &t) else {
1266 return Self::empty();
1267 };
1268 let mut table: HashMap<Vec<Val>, Vec<Row>> = HashMap::new();
1269 for r in rel {
1270 let key: Vec<Val> = shared.iter().map(|&s| r[s].clone().unwrap()).collect();
1271 table.entry(key).or_default().push(r);
1272 }
1273 let fill = slots.clone();
1274 stream = Box::new(stream.flat_map(move |base| {
1275 let key: Vec<Val> = shared.iter().map(|&s| base[s].clone().unwrap()).collect();
1276 let out: Vec<Row> = table
1277 .get(&key)
1278 .map(|rs| {
1279 rs.iter()
1280 .map(|r| {
1281 let mut row = base.clone();
1282 for &s in &fill {
1283 row[s] = r[s].clone();
1284 }
1285 row
1286 })
1287 .collect()
1288 })
1289 .unwrap_or_default();
1290 out.into_iter()
1291 }));
1292 }
1293 for s in slots {
1294 if !bound.contains(&s) {
1295 bound.push(s);
1296 }
1297 }
1298 }
1299 Self::passthrough(stream)
1300 }
1301
1302 fn empty() -> Self {
1304 BgpSolutions {
1305 scan: None,
1306 prefix: Vec::new(),
1307 buckets: HashMap::new(),
1308 shared: Vec::new(),
1309 cartesian: false,
1310 seed_only: false,
1311 cur_scan: None,
1312 matches: Vec::new(),
1313 mi: 0,
1314 }
1315 }
1316
1317 pub(crate) fn new(ctx: &'q Ctx<'q>, index: &'q GraphIndex, patterns: &[TriplePattern]) -> Self {
1318 if patterns.is_empty() {
1320 return BgpSolutions {
1321 scan: Some(Box::new(std::iter::once(ctx.slots.empty_row()))),
1322 prefix: vec![ctx.slots.empty_row()],
1323 buckets: HashMap::new(),
1324 shared: Vec::new(),
1325 cartesian: true,
1326 seed_only: true,
1327 cur_scan: None,
1328 matches: Vec::new(),
1329 mi: 0,
1330 };
1331 }
1332 let Some(lowered) = lower(patterns, ctx) else {
1334 return Self::empty();
1335 };
1336 let order = selectivity_order(ctx, index, &lowered);
1339 let scan_est = tile_estimates(ctx, index, &lowered, &std::collections::HashSet::new());
1340 if order.len() >= 2 && scan_est[order[0]] > limits().stream_seed {
1343 return Self::streaming(ctx, index, &lowered, &order, &scan_est);
1344 }
1345 let (&last_i, prefix_is) = order.split_last().unwrap();
1346 let prefix_pats: Vec<TriplePattern> =
1347 prefix_is.iter().map(|&i| patterns[i].clone()).collect();
1348 let prefix = eval_bgp_rows(ctx, index, &prefix_pats);
1349 if prefix.is_empty() {
1350 return Self::empty();
1351 }
1352
1353 let shared: Vec<usize> = pattern_slots(&lowered[last_i])
1357 .into_iter()
1358 .filter(|&s| prefix[0][s].is_some())
1359 .collect();
1360
1361 let class_probe = !shared.is_empty()
1370 && scan_est[last_i] >= limits().probe_floor
1371 && (prefix.len() as f64) <= scan_est[last_i]
1372 && is_unbound_class(ctx, &lowered[last_i], &shared);
1373 if !shared.is_empty() && (probe_beats_scan(prefix.len(), scan_est[last_i]) || class_probe) {
1374 let t = lowered[last_i];
1375 if index.is_remote() {
1376 prefetch_rows_for(ctx, index, t, &prefix);
1377 }
1378 let stream = prefix
1379 .into_iter()
1380 .flat_map(move |base| probe_rows(ctx, index, t, base));
1381 return Self::passthrough(Box::new(stream));
1382 }
1383
1384 let Some(scan) = scan_rows(ctx, index, lowered[last_i]) else {
1386 return Self::empty();
1387 };
1388 let cartesian = shared.is_empty();
1389 let seed_only = prefix.len() == 1 && prefix[0].iter().all(Option::is_none);
1392 let mut buckets: HashMap<Vec<Val>, Vec<usize>> = HashMap::new();
1393 if !cartesian {
1394 for (i, r) in prefix.iter().enumerate() {
1395 let key: Vec<Val> = shared.iter().map(|&s| r[s].clone().unwrap()).collect();
1396 buckets.entry(key).or_default().push(i);
1397 }
1398 }
1399 BgpSolutions {
1400 scan: Some(Box::new(scan)),
1401 prefix,
1402 buckets,
1403 shared,
1404 cartesian,
1405 seed_only,
1406 cur_scan: None,
1407 matches: Vec::new(),
1408 mi: 0,
1409 }
1410 }
1411}
1412
1413impl Iterator for BgpSolutions<'_> {
1414 type Item = Row;
1415
1416 fn next(&mut self) -> Option<Row> {
1417 if self.seed_only {
1419 return self.scan.as_mut()?.next();
1420 }
1421 loop {
1422 if self.mi < self.matches.len() {
1424 let pi = self.matches[self.mi];
1425 self.mi += 1;
1426 let mut merged = self.prefix[pi].clone();
1427 for (slot, v) in self.cur_scan.as_ref().unwrap().iter().enumerate() {
1428 if v.is_some() {
1429 merged[slot] = v.clone();
1430 }
1431 }
1432 return Some(merged);
1433 }
1434 let s = self.scan.as_mut()?.next()?;
1437 self.matches.clear();
1438 if self.cartesian {
1439 self.matches.extend(0..self.prefix.len());
1440 } else {
1441 let key: Vec<Val> = self.shared.iter().map(|&i| s[i].clone().unwrap()).collect();
1442 if let Some(c) = self.buckets.get(&key) {
1443 self.matches.extend_from_slice(c);
1444 }
1445 }
1446 self.mi = 0;
1447 self.cur_scan = Some(s);
1448 }
1449 }
1450}
1451
1452fn prefetch_rows_for(
1455 ctx: &Ctx,
1456 index: &GraphIndex,
1457 t: (SlotTerm, SlotTerm, SlotTerm),
1458 rows: &[Row],
1459) {
1460 let pats: Vec<Pattern> = rows
1461 .iter()
1462 .filter_map(|base| {
1463 match (
1464 probe_subject(ctx, &t.0, base),
1465 probe_predicate(ctx, &t.1, base),
1466 probe_object(ctx, &t.2, base),
1467 ) {
1468 (Some(s), Some(p), Some(o)) => Some((s, p, o)),
1469 _ => None,
1470 }
1471 })
1472 .collect();
1473 index.prefetch_probe_tiles(&pats);
1474}
1475
1476#[allow(clippy::type_complexity)]
1484fn merge_stream<'q>(
1485 ctx: &'q Ctx<'q>,
1486 index: &'q GraphIndex,
1487 ta: (SlotTerm, SlotTerm, SlotTerm),
1488 tb: (SlotTerm, SlotTerm, SlotTerm),
1489) -> Option<(Box<dyn Iterator<Item = Row> + 'q>, Vec<usize>)> {
1490 let sa = pattern_slots(&ta);
1491 let sb = pattern_slots(&tb);
1492 let shared: Vec<usize> = sa.iter().copied().filter(|s| sb.contains(s)).collect();
1493 let [v] = shared[..] else {
1494 return None;
1495 };
1496 if !matches!(ta.0, SlotTerm::Var(i) if i == v) || !matches!(tb.0, SlotTerm::Var(i) if i == v) {
1497 return None;
1498 }
1499 let dict = ctx.rete.dictionary();
1500 let lower = |t: &(SlotTerm, SlotTerm, SlotTerm)| -> Option<Pattern> {
1501 Some((
1502 const_subject(&t.0, dict)?,
1503 const_predicate(&t.1)?,
1504 const_object(&t.2, dict)?,
1505 ))
1506 };
1507 let side = |t: (SlotTerm, SlotTerm, SlotTerm)| -> Option<KeyedRows<'q>> {
1508 let it = index.scan_iter_sorted_on(lower(&t)?, 0)?;
1509 let keyed: Box<dyn Iterator<Item = (u32, Row)> + 'q> =
1510 Box::new(it.filter_map(move |tri| triple_row(ctx, &t, tri).map(|r| (tri.0, r))));
1511 Some(keyed.peekable())
1512 };
1513 let a = side(ta)?;
1514 let b = side(tb)?;
1515 let mut slots = sa;
1516 for s in sb {
1517 if !slots.contains(&s) {
1518 slots.push(s);
1519 }
1520 }
1521 Some((
1522 Box::new(MergeStream {
1523 a,
1524 b,
1525 ra: Vec::new(),
1526 rb: Vec::new(),
1527 i: 0,
1528 j: 0,
1529 }),
1530 slots,
1531 ))
1532}
1533
1534type KeyedRows<'q> = std::iter::Peekable<Box<dyn Iterator<Item = (u32, Row)> + 'q>>;
1535
1536struct MergeStream<'q> {
1539 a: KeyedRows<'q>,
1540 b: KeyedRows<'q>,
1541 ra: Vec<Row>,
1542 rb: Vec<Row>,
1543 i: usize,
1544 j: usize,
1545}
1546
1547impl Iterator for MergeStream<'_> {
1548 type Item = Row;
1549
1550 fn next(&mut self) -> Option<Row> {
1551 loop {
1552 while self.i < self.ra.len() {
1553 if self.j < self.rb.len() {
1554 let j = self.j;
1555 self.j += 1;
1556 if let Some(m) = crate::row::merge_rows(&self.ra[self.i], &self.rb[j]) {
1557 return Some(m);
1558 }
1559 } else {
1560 self.i += 1;
1561 self.j = 0;
1562 }
1563 }
1564 let ka = self.a.peek()?.0;
1565 let kb = self.b.peek()?.0;
1566 match ka.cmp(&kb) {
1567 std::cmp::Ordering::Less => {
1568 self.a.next();
1569 }
1570 std::cmp::Ordering::Greater => {
1571 self.b.next();
1572 }
1573 std::cmp::Ordering::Equal => {
1574 self.ra.clear();
1575 self.rb.clear();
1576 while let Some((_, r)) = self.a.next_if(|(k, _)| *k == ka) {
1577 self.ra.push(r);
1578 }
1579 while let Some((_, r)) = self.b.next_if(|(k, _)| *k == ka) {
1580 self.rb.push(r);
1581 }
1582 self.i = 0;
1583 self.j = 0;
1584 }
1585 }
1586 }
1587 }
1588}
1589
1590fn probe_subject(ctx: &Ctx, t: &SlotTerm, base: &Row) -> Option<Option<u32>> {
1604 let dict = ctx.rete.dictionary();
1605 match t {
1606 SlotTerm::Node(n) => dict.node_as_subject_id(*n).map(Some),
1607 SlotTerm::Pred(_) => None,
1608 SlotTerm::Var(i) => match &base[*i] {
1609 None => Some(None),
1610 Some(Val::Id(v)) if *v >= 0 => dict.node_as_subject_id(*v as u32).map(Some),
1611 Some(_) => None,
1613 },
1614 }
1615}
1616
1617fn probe_object(ctx: &Ctx, t: &SlotTerm, base: &Row) -> Option<Option<u32>> {
1619 let dict = ctx.rete.dictionary();
1620 match t {
1621 SlotTerm::Node(n) => dict.node_as_object_id(*n).map(Some),
1622 SlotTerm::Pred(_) => None,
1623 SlotTerm::Var(i) => match &base[*i] {
1624 None => Some(None),
1625 Some(Val::Id(v)) if *v >= 0 => dict.node_as_object_id(*v as u32).map(Some),
1626 Some(_) => None,
1627 },
1628 }
1629}
1630
1631fn probe_predicate(ctx: &Ctx, t: &SlotTerm, base: &Row) -> Option<Option<u32>> {
1633 let dict = ctx.rete.dictionary();
1634 match t {
1635 SlotTerm::Pred(p) => Some(Some(*p)),
1636 SlotTerm::Node(_) => None,
1637 SlotTerm::Var(i) => match &base[*i] {
1638 None => Some(None),
1639 Some(Val::Id(v)) if *v < 0 => Some(Some((-v - 1) as u32)),
1640 Some(Val::Id(v)) => ctx
1642 .resolver
1643 .term(*v)
1644 .and_then(|t| dict.predicate_id(&t))
1645 .map(Some),
1646 Some(Val::Str(_)) => None,
1647 },
1648 }
1649}
1650
1651fn probe_rows<'q>(
1655 ctx: &'q Ctx<'q>,
1656 index: &'q GraphIndex,
1657 t: (SlotTerm, SlotTerm, SlotTerm),
1658 base: Row,
1659) -> Box<dyn Iterator<Item = Row> + 'q> {
1660 let (Some(sid), Some(pid), Some(oid)) = (
1661 probe_subject(ctx, &t.0, &base),
1662 probe_predicate(ctx, &t.1, &base),
1663 probe_object(ctx, &t.2, &base),
1664 ) else {
1665 return Box::new(std::iter::empty());
1666 };
1667 let dict = ctx.rete.dictionary();
1668 Box::new(
1669 index
1670 .scan_iter((sid, pid, oid))
1671 .filter_map(move |(s_id, p_id, o_id)| {
1672 let s_val = dict.subject_node(s_id) as i64;
1673 let p_val = pred_val(ctx, &t.1, p_id);
1674 let o_val = dict.object_node(o_id) as i64;
1675 let mut row = base.clone();
1676 for (term, val) in [(&t.0, s_val), (&t.1, p_val), (&t.2, o_val)] {
1677 if let SlotTerm::Var(i) = term {
1678 match row[*i] {
1679 Some(Val::Id(existing)) if existing != val => return None,
1680 Some(Val::Id(_)) => {}
1681 Some(Val::Str(_)) => return None,
1682 None => row[*i] = Some(Val::Id(val)),
1683 }
1684 }
1685 }
1686 Some(row)
1687 }),
1688 )
1689}
1690
1691pub(crate) struct ProbePlan {
1693 pats: Vec<(SlotTerm, SlotTerm, SlotTerm)>,
1694}
1695
1696impl ProbePlan {
1697 pub(crate) fn new(
1701 ctx: &Ctx,
1702 index: &GraphIndex,
1703 patterns: &[TriplePattern],
1704 seed_mask: &[bool],
1705 ) -> Option<Self> {
1706 let lowered = lower(patterns, ctx)?;
1707 let seed: std::collections::HashSet<usize> = seed_mask
1708 .iter()
1709 .enumerate()
1710 .filter_map(|(i, b)| b.then_some(i))
1711 .collect();
1712 let order = selectivity_order_seeded(ctx, index, &lowered, &seed);
1713 Some(ProbePlan {
1714 pats: order.into_iter().map(|i| lowered[i]).collect(),
1715 })
1716 }
1717}
1718
1719pub(crate) struct ProbeJoin<'q> {
1722 ctx: &'q Ctx<'q>,
1723 index: &'q GraphIndex,
1724 pats: Vec<(SlotTerm, SlotTerm, SlotTerm)>,
1725 stack: Vec<Box<dyn Iterator<Item = Row> + 'q>>,
1726}
1727
1728impl<'q> ProbeJoin<'q> {
1729 pub(crate) fn new(
1732 ctx: &'q Ctx<'q>,
1733 index: &'q GraphIndex,
1734 patterns: &[TriplePattern],
1735 ) -> Option<Self> {
1736 let plan = ProbePlan::new(ctx, index, patterns, &vec![false; ctx.slots.len()])?;
1737 Some(Self::from_plan(ctx, index, &plan, ctx.slots.empty_row()))
1738 }
1739
1740 pub(crate) fn from_plan(
1742 ctx: &'q Ctx<'q>,
1743 index: &'q GraphIndex,
1744 plan: &ProbePlan,
1745 seed: Row,
1746 ) -> Self {
1747 let pats = plan.pats.clone();
1748 let first = probe_rows(ctx, index, pats[0], seed);
1749 ProbeJoin {
1750 ctx,
1751 index,
1752 pats,
1753 stack: vec![first],
1754 }
1755 }
1756}
1757
1758impl Iterator for ProbeJoin<'_> {
1759 type Item = Row;
1760
1761 fn next(&mut self) -> Option<Row> {
1762 loop {
1763 let depth = self.stack.len();
1764 match self.stack.last_mut()?.next() {
1765 Some(row) => {
1766 if depth == self.pats.len() {
1767 return Some(row);
1768 }
1769 let it = probe_rows(self.ctx, self.index, self.pats[depth], row);
1770 self.stack.push(it);
1771 }
1772 None => {
1773 self.stack.pop();
1774 if self.stack.is_empty() {
1775 return None;
1776 }
1777 }
1778 }
1779 }
1780 }
1781}
1782
1783fn hash_join(
1787 left: Vec<Row>,
1788 left_bound: &[usize],
1789 right: Vec<Row>,
1790 right_slots: &[usize],
1791) -> Vec<Row> {
1792 if left.len() == 1 && left_bound.is_empty() {
1794 return right;
1795 }
1796 if left.is_empty() || right.is_empty() {
1797 return Vec::new();
1798 }
1799 let shared: Vec<usize> = right_slots
1800 .iter()
1801 .copied()
1802 .filter(|s| left_bound.contains(s))
1803 .collect();
1804 let fill = |l: &Row, r: &Row| -> Row {
1805 let mut out = l.clone();
1806 for &s in right_slots {
1807 out[s] = r[s].clone();
1808 }
1809 out
1810 };
1811 if shared.is_empty() {
1812 let mut out = Vec::with_capacity(left.len() * right.len());
1816 for l in &left {
1817 for r in &right {
1818 out.push(fill(l, r));
1819 }
1820 }
1821 return out;
1822 }
1823 let key_of = |b: &Row| -> Vec<Val> { shared.iter().map(|&s| b[s].clone().unwrap()).collect() };
1824 let mut out = Vec::new();
1831 if right.len() <= left.len() {
1832 let mut buckets: HashMap<Vec<Val>, Vec<Row>> = HashMap::new();
1833 for r in right {
1834 buckets.entry(key_of(&r)).or_default().push(r);
1835 }
1836 for l in &left {
1837 if let Some(rs) = buckets.get(&key_of(l)) {
1838 for r in rs {
1839 out.push(fill(l, r));
1840 }
1841 }
1842 }
1843 } else {
1844 let mut buckets: HashMap<Vec<Val>, Vec<Row>> = HashMap::new();
1846 for l in left {
1847 buckets.entry(key_of(&l)).or_default().push(l);
1848 }
1849 for r in &right {
1850 if let Some(ls) = buckets.get(&key_of(r)) {
1851 for l in ls {
1852 let mut row = r.clone();
1853 for &s in left_bound {
1854 if let Some(v) = &l[s] {
1855 row[s] = Some(v.clone());
1856 }
1857 }
1858 out.push(row);
1859 }
1860 }
1861 }
1862 }
1863 out
1864}
1865
1866fn const_subject(t: &SlotTerm, d: &crate::Dictionary) -> Option<Option<u32>> {
1871 match t {
1872 SlotTerm::Node(n) => d.node_as_subject_id(*n).map(Some),
1873 SlotTerm::Pred(_) => None, SlotTerm::Var(_) => Some(None),
1875 }
1876}
1877
1878fn const_object(t: &SlotTerm, d: &crate::Dictionary) -> Option<Option<u32>> {
1879 match t {
1880 SlotTerm::Node(n) => d.node_as_object_id(*n).map(Some),
1881 SlotTerm::Pred(_) => None,
1882 SlotTerm::Var(_) => Some(None),
1883 }
1884}
1885
1886fn const_predicate(t: &SlotTerm) -> Option<Option<u32>> {
1887 match t {
1888 SlotTerm::Pred(p) => Some(Some(*p)),
1889 SlotTerm::Node(_) => None, SlotTerm::Var(_) => Some(None),
1891 }
1892}
1893
1894#[cfg(test)]
1895mod tests {
1896 use super::*;
1897 use crate::dictionary::DictionaryBuilder;
1898 use crate::file::write_file_v05 as write_file;
1899 use crate::index::GraphIndexBuilder;
1900
1901 fn rete_from(triples: &[(&str, &str, &str)]) -> Vec<u8> {
1902 let mut db = DictionaryBuilder::new();
1903 for (s, p, o) in triples {
1904 db.observe(s, p, o);
1905 }
1906 let dict = db.build();
1907 let mut ib = GraphIndexBuilder::new();
1908 for (s, p, o) in triples {
1909 ib.push(dict.encode(s, p, o).unwrap());
1910 }
1911 write_file(&dict, &ib.build(), false, &[], 0)
1912 }
1913
1914 fn pat(s: &str, p: &str, o: &str) -> TriplePattern {
1915 TriplePattern {
1916 s: PatternTerm::parse(s),
1917 p: PatternTerm::parse(p),
1918 o: PatternTerm::parse(o),
1919 }
1920 }
1921
1922 #[test]
1931 fn scan_bytes_route_a_literal_by_its_object_id() {
1932 let value = "<http://ex/value>";
1933 let owned: Vec<(String, String)> = (0..400)
1934 .map(|i| (format!("<http://ex/s{i:03}>"), format!("\"lit {i:03}\"")))
1935 .collect();
1936 let mut db = DictionaryBuilder::new();
1937 for (s, o) in &owned {
1938 db.observe(s, value, o);
1939 }
1940 let dict = db.build();
1941 let mut ib = GraphIndexBuilder::new().with_tile_budget(256);
1942 for (s, o) in &owned {
1943 ib.push(dict.encode(s, value, o).unwrap());
1944 }
1945 let rete = Rete::open(&write_file(&dict, &ib.build(), false, &[], 0)).unwrap();
1946 let index = rete.default_index();
1947 let whole = routed_scan_bytes(index, (None, rete.dictionary().predicate_id(value), None));
1949 for (_, lit) in owned.iter().step_by(57) {
1950 let q = [pat("?s", value, lit)];
1951 let mut slots = Slots::new();
1952 collect_pattern_slots(&q, &mut slots);
1953 let ctx = Ctx::new(&rete, slots);
1954 let got = bgp_min_scan_bytes(&ctx, index, &q).unwrap();
1955 assert!(got > 0, "{lit} routed to no tile");
1956 assert!(
1957 got < whole,
1958 "{lit} routed to the whole predicate ({got} of {whole})"
1959 );
1960 }
1961 }
1962
1963 #[test]
1969 fn plan_gates_do_not_move_with_the_compression() {
1970 let mut owned: Vec<(String, String, String)> = Vec::new();
1971 for i in 0..3000 {
1972 let s = format!("<http://ex/s{i}>");
1973 owned.push((
1974 s.clone(),
1975 "<http://ex/kind>".into(),
1976 format!("<http://ex/K{}>", i % 7),
1977 ));
1978 owned.push((s.clone(), "<http://ex/value>".into(), format!("\"v {i}\"")));
1979 if i % 50 == 0 {
1980 owned.push((s, "<http://ex/rare>".into(), "<http://ex/R>".into()));
1981 }
1982 }
1983 let mut db = DictionaryBuilder::new();
1984 for (s, p, o) in &owned {
1985 db.observe(s, p, o);
1986 }
1987 let dict = db.build();
1988 let mut ib = GraphIndexBuilder::new().with_tile_budget(1024);
1989 for (s, p, o) in &owned {
1990 ib.push(dict.encode(s, p, o).unwrap());
1991 }
1992 let bytes = write_file(&dict, &ib.build(), false, &[], 0);
1993 let leaked: &'static [u8] = Box::leak(bytes.clone().into_boxed_slice());
1994 let resident = Rete::open(&bytes).unwrap();
1995 let lazy = Rete::open_ranged_lazy(crate::reader::SliceReader::new(leaked)).unwrap();
1996 let shapes: Vec<Vec<TriplePattern>> = vec![
1997 vec![
1998 pat("?s", "<http://ex/rare>", "<http://ex/R>"),
1999 pat("?s", "<http://ex/value>", "?v"),
2000 ],
2001 vec![
2002 pat("?s", "<http://ex/kind>", "<http://ex/K3>"),
2003 pat("?s", "<http://ex/value>", "?v"),
2004 ],
2005 vec![
2006 pat("?s", "<http://ex/kind>", "?k"),
2007 pat("?s", "<http://ex/value>", "?v"),
2008 ],
2009 vec![
2010 pat("?s", "<http://ex/value>", "\"v 7\""),
2011 pat("?s", "<http://ex/kind>", "?k"),
2012 ],
2013 ];
2014 let decisions = |rete: &Rete| -> Vec<(Vec<u64>, bool, Option<u64>)> {
2015 shapes
2016 .iter()
2017 .map(|q| {
2018 let mut slots = Slots::new();
2019 collect_pattern_slots(q, &mut slots);
2020 let ctx = Ctx::new(rete, slots);
2021 let index = rete.default_index();
2022 let lowered = lower(q, &ctx).unwrap();
2023 let est = tile_estimates(&ctx, index, &lowered, &Default::default());
2024 let lopsided = merge_seed_lopsided(&ctx, index, &lowered[0], &lowered[1]);
2025 let min = bgp_min_estimate(&ctx, index, q).map(|e| e as u64);
2026 (est.iter().map(|e| *e as u64).collect(), lopsided, min)
2027 })
2028 .collect()
2029 };
2030 assert_eq!(decisions(&resident), decisions(&lazy));
2031 #[cfg(feature = "compression")]
2033 {
2034 let bytes_of = |rete: &Rete| {
2035 let q = &shapes[2];
2036 let mut slots = Slots::new();
2037 collect_pattern_slots(q, &mut slots);
2038 let ctx = Ctx::new(rete, slots);
2039 bgp_min_scan_bytes(&ctx, rete.default_index(), q)
2040 };
2041 assert_ne!(bytes_of(&resident), bytes_of(&lazy));
2042 }
2043 }
2044
2045 #[test]
2050 fn every_join_path_returns_the_same_solutions() {
2051 let mut owned: Vec<(String, String, String)> = Vec::new();
2052 let mut x: u64 = 7;
2053 let mut next = |m: u64| {
2054 x = x
2055 .wrapping_mul(6364136223846793005)
2056 .wrapping_add(1442695040888963407);
2057 (x >> 33) % m
2058 };
2059 for i in 0..60 {
2060 let s = format!("s{i}");
2061 owned.push((s.clone(), "a".into(), format!("C{}", next(3))));
2062 for p in 0..4 {
2063 for _ in 0..next(3) {
2064 owned.push((s.clone(), format!("p{p}"), format!("s{}", next(60))));
2065 }
2066 }
2067 }
2068 let triples: Vec<(&str, &str, &str)> = owned
2069 .iter()
2070 .map(|(s, p, o)| (s.as_str(), p.as_str(), o.as_str()))
2071 .collect();
2072 let bytes = rete_from(&triples);
2073 let rete = Rete::open(&bytes).unwrap();
2074 let queries: Vec<Vec<TriplePattern>> = vec![
2075 vec![pat("?s", "p0", "?a"), pat("?s", "p1", "?b")],
2076 vec![pat("?s", "a", "C1"), pat("?s", "p0", "?a")],
2077 vec![pat("?s", "p0", "?a"), pat("?a", "p2", "?b")],
2078 vec![pat("s5", "p0", "?a"), pat("?a", "?p", "?o")],
2079 vec![pat("?s", "p0", "?x"), pat("?s", "p1", "?x")],
2080 vec![
2081 pat("?s", "p3", "?o"),
2082 pat("?o", "a", "C0"),
2083 pat("?s", "p1", "?z"),
2084 ],
2085 vec![
2086 pat("?s", "p0", "?a"),
2087 pat("?s", "p1", "?b"),
2088 pat("?b", "p2", "?c"),
2089 ],
2090 ];
2091 let run = |limits: Option<PlanLimits>| -> Vec<Vec<Vec<(String, String)>>> {
2092 TEST_LIMITS.with(|c| c.set(limits));
2093 let out = queries
2094 .iter()
2095 .map(|q| {
2096 let mut slots = crate::row::Slots::new();
2097 collect_pattern_slots(q, &mut slots);
2098 let ctx = Ctx::new(&rete, slots);
2099 let mut rows: Vec<Vec<(String, String)>> =
2100 BgpSolutions::new(&ctx, rete.default_index(), q)
2101 .map(|r| row_to_binding(&ctx, &r).into_iter().collect())
2102 .collect();
2103 rows.sort();
2104 rows
2105 })
2106 .collect();
2107 TEST_LIMITS.with(|c| c.set(None));
2108 out
2109 };
2110 let reference = run(None);
2111 assert!(reference.iter().all(|r| !r.is_empty()), "{reference:?}");
2112 for (name, l) in [
2113 (
2114 "probe everything",
2115 PlanLimits {
2116 probe_floor: 0.0,
2117 stream_seed: f64::MAX,
2118 hash_build: f64::MAX,
2119 },
2120 ),
2121 (
2122 "stream + hash",
2123 PlanLimits {
2124 probe_floor: f64::MAX,
2125 stream_seed: 0.0,
2126 hash_build: f64::MAX,
2127 },
2128 ),
2129 (
2130 "stream + probe",
2131 PlanLimits {
2132 probe_floor: f64::MAX,
2133 stream_seed: 0.0,
2134 hash_build: 0.0,
2135 },
2136 ),
2137 (
2138 "stream, all paths",
2139 PlanLimits {
2140 probe_floor: 0.0,
2141 stream_seed: 0.0,
2142 hash_build: 30.0,
2143 },
2144 ),
2145 ] {
2146 assert_eq!(run(Some(l)), reference, "{name}");
2147 }
2148 }
2149
2150 #[test]
2151 fn single_pattern_binds_variable() {
2152 let bytes = rete_from(&[("Alice", "knows", "Bob"), ("Bob", "knows", "Carol")]);
2153 let rete = Rete::open(&bytes).unwrap();
2154 let sols = eval_bgp(&rete, &[pat("Alice", "knows", "?y")]);
2155 assert_eq!(sols.len(), 1);
2156 assert_eq!(sols[0]["y"], "Bob");
2157 }
2158
2159 #[test]
2160 fn two_hop_join_on_shared_variable() {
2161 let bytes = rete_from(&[
2163 ("Alice", "knows", "Bob"),
2164 ("Bob", "knows", "Carol"),
2165 ("Carol", "knows", "Dave"),
2166 ("Alice", "knows", "Eve"), ]);
2168 let rete = Rete::open(&bytes).unwrap();
2169 let sols = eval_bgp(&rete, &[pat("?x", "knows", "?y"), pat("?y", "knows", "?z")]);
2170 let mut got: Vec<_> = sols
2172 .iter()
2173 .map(|b| (b["x"].clone(), b["y"].clone(), b["z"].clone()))
2174 .collect();
2175 got.sort();
2176 assert_eq!(
2177 got,
2178 vec![
2179 ("Alice".into(), "Bob".into(), "Carol".into()),
2180 ("Bob".into(), "Carol".into(), "Dave".into()),
2181 ]
2182 );
2183 }
2184
2185 #[test]
2186 fn repeated_variable_within_pattern() {
2187 let bytes = rete_from(&[("Alice", "knows", "Bob"), ("Bob", "knows", "Bob")]);
2189 let rete = Rete::open(&bytes).unwrap();
2190 let sols = eval_bgp(&rete, &[pat("?x", "knows", "?x")]);
2191 assert_eq!(sols.len(), 1);
2192 assert_eq!(sols[0]["x"], "Bob");
2193 }
2194
2195 #[test]
2196 fn no_solutions_yields_empty() {
2197 let bytes = rete_from(&[("Alice", "knows", "Bob")]);
2198 let rete = Rete::open(&bytes).unwrap();
2199 assert!(eval_bgp(&rete, &[pat("Alice", "likes", "?y")]).is_empty());
2200 }
2201
2202 #[test]
2209 fn snowflake_type_and_adjacency_matches_reference() {
2210 const TYPE: &str = "<http://www.w3.org/1999/02/22-rdf-syntax-ns#type>";
2211 let mut triples: Vec<(String, String, String)> = Vec::new();
2212 for i in 0..60 {
2214 let s = format!("S{i}");
2215 triples.push((s.clone(), TYPE.into(), "Student".into()));
2216 triples.push((s.clone(), "takesCourse".into(), format!("c{}", i % 5)));
2217 triples.push((s, "takesCourse".into(), format!("c{}", (i + 1) % 5)));
2218 }
2219 for j in 0..5 {
2221 triples.push((format!("c{j}"), TYPE.into(), "Course".into()));
2222 }
2223 triples.push(("Prof".into(), "teacherOf".into(), "c0".into()));
2224 triples.push(("Prof".into(), "teacherOf".into(), "c2".into()));
2225
2226 let refs: Vec<(&str, &str, &str)> = triples
2227 .iter()
2228 .map(|(s, p, o)| (s.as_str(), p.as_str(), o.as_str()))
2229 .collect();
2230 let bytes = rete_from(&refs);
2231 let rete = Rete::open(&bytes).unwrap();
2232
2233 let got: std::collections::BTreeSet<(String, String)> = eval_bgp(
2234 &rete,
2235 &[
2236 pat("?x", TYPE, "Student"),
2237 pat("?y", TYPE, "Course"),
2238 pat("?x", "takesCourse", "?y"),
2239 pat("Prof", "teacherOf", "?y"),
2240 ],
2241 )
2242 .into_iter()
2243 .map(|b| (b["x"].clone(), b["y"].clone()))
2244 .collect();
2245
2246 let taught = ["c0", "c2"];
2248 let mut want = std::collections::BTreeSet::new();
2249 for i in 0..60 {
2250 for c in [i % 5, (i + 1) % 5] {
2251 let course = format!("c{c}");
2252 if taught.contains(&course.as_str()) {
2253 want.insert((format!("S{i}"), course));
2254 }
2255 }
2256 }
2257 assert_eq!(got, want, "snowflake join must match the brute-force set");
2258 assert!(!want.is_empty(), "sanity: the reference set is non-empty");
2259 }
2260
2261 #[test]
2266 fn merge_join_subject_star_cross_product() {
2267 let bytes = rete_from(&[
2268 ("x", "a", "a1"),
2269 ("x", "a", "a2"),
2270 ("x", "b", "b1"),
2271 ("x", "b", "b2"),
2272 ("y", "a", "a3"),
2273 ("y", "b", "b3"),
2274 ("z", "a", "a4"), ]);
2276 let rete = Rete::open(&bytes).unwrap();
2277 let mut got: Vec<(String, String, String)> =
2278 eval_bgp(&rete, &[pat("?x", "a", "?av"), pat("?x", "b", "?bv")])
2279 .iter()
2280 .map(|m| (m["x"].clone(), m["av"].clone(), m["bv"].clone()))
2281 .collect();
2282 got.sort();
2283 let mut want: Vec<(String, String, String)> = [
2284 ("x", "a1", "b1"),
2285 ("x", "a1", "b2"),
2286 ("x", "a2", "b1"),
2287 ("x", "a2", "b2"),
2288 ("y", "a3", "b3"),
2289 ]
2290 .into_iter()
2291 .map(|(a, b, c)| (a.to_string(), b.to_string(), c.to_string()))
2292 .collect();
2293 want.sort();
2294 assert_eq!(got, want, "same-role merge must equal the brute-force join");
2295 }
2296}