use std::collections::{BTreeMap, HashMap};
use crate::file::Rete;
use crate::index::{GraphIndex, Pattern, Tile};
use crate::row::{Ctx, Row, Slots, Val};
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PatternTerm {
Var(String),
Const(String),
}
impl PatternTerm {
pub fn parse(token: &str) -> Self {
if let Some(name) = token.strip_prefix('?') {
PatternTerm::Var(name.to_string())
} else {
PatternTerm::Const(token.to_string())
}
}
}
#[derive(Debug, Clone)]
pub struct TriplePattern {
pub s: PatternTerm,
pub p: PatternTerm,
pub o: PatternTerm,
}
pub type Binding = BTreeMap<String, String>;
#[derive(Clone, Copy)]
enum SlotTerm {
Var(usize),
Node(u32),
Pred(u32),
}
fn pred_tag(p: u32) -> i64 {
-(p as i64) - 1
}
pub(crate) fn collect_pattern_slots(patterns: &[TriplePattern], slots: &mut Slots) {
for p in patterns {
for t in [&p.s, &p.p, &p.o] {
if let PatternTerm::Var(v) = t {
slots.add(v);
}
}
}
}
fn lower(patterns: &[TriplePattern], ctx: &Ctx) -> Option<Vec<(SlotTerm, SlotTerm, SlotTerm)>> {
let dict = ctx.rete.dictionary();
let node = |t: &PatternTerm| -> Option<SlotTerm> {
match t {
PatternTerm::Var(v) => ctx.slots.slot(v).map(SlotTerm::Var),
PatternTerm::Const(c) => dict.node_of_term(c).map(SlotTerm::Node),
}
};
let pred = |t: &PatternTerm| -> Option<SlotTerm> {
match t {
PatternTerm::Var(v) => ctx.slots.slot(v).map(SlotTerm::Var),
PatternTerm::Const(c) => dict.predicate_id(c).map(SlotTerm::Pred),
}
};
let mut lowered = Vec::with_capacity(patterns.len());
for p in patterns {
lowered.push((node(&p.s)?, pred(&p.p)?, node(&p.o)?));
}
Some(lowered)
}
pub fn eval_bgp(rete: &Rete, patterns: &[TriplePattern]) -> Vec<Binding> {
let mut slots = Slots::new();
collect_pattern_slots(patterns, &mut slots);
let ctx = Ctx::new(rete, slots);
eval_bgp_rows(&ctx, rete.default_index(), patterns)
.into_iter()
.map(|row| row_to_binding(&ctx, &row))
.collect()
}
pub(crate) fn row_to_binding(ctx: &Ctx, row: &Row) -> Binding {
let mut b = Binding::new();
for (i, v) in row.iter().enumerate() {
if let Some(val) = v {
if let Some(t) = ctx.resolver.str_once(val) {
b.insert(ctx.slots.name(i).to_string(), t);
}
}
}
b
}
pub(crate) fn eval_bgp_rows(ctx: &Ctx, index: &GraphIndex, patterns: &[TriplePattern]) -> Vec<Row> {
if patterns.is_empty() {
return vec![ctx.slots.empty_row()];
}
let Some(lowered) = lower(patterns, ctx) else {
return Vec::new();
};
let order = selectivity_order(ctx, &lowered);
let mut rows: Vec<Row> = vec![ctx.slots.empty_row()];
let mut bound: Vec<usize> = Vec::new();
let mut merged: [bool; 2] = [false, false];
if order.len() >= 2 {
if let Some((rel, slots)) =
try_merge_join(ctx, index, &lowered[order[0]], &lowered[order[1]])
{
rows = rel;
bound = slots;
merged = [true, true];
}
}
for (k, &idx) in order.iter().enumerate() {
if k < 2 && merged[k] {
continue; }
let t = lowered[idx];
let shares_bound = pattern_slots(&t).iter().any(|s| bound.contains(s));
let do_probe = !bound.is_empty()
&& shares_bound
&& !rows.is_empty()
&& if index.is_remote() {
rows.len() <= remote_probe_max(index)
&& (rows.len() <= REMOTE_PROBE_MIN || !pattern_is_selective(&t))
} else {
rows.len() <= BGP_PROBE_THRESHOLD
};
if do_probe {
let taken = std::mem::take(&mut rows);
if index.is_remote() {
let pats: Vec<Pattern> = taken
.iter()
.filter_map(|base| {
match (
probe_subject(ctx, &t.0, base),
probe_predicate(ctx, &t.1, base),
probe_object(ctx, &t.2, base),
) {
(Some(s), Some(p), Some(o)) => Some((s, p, o)),
_ => None,
}
})
.collect();
index.prefetch_probe_tiles(&pats);
}
let mut next: Vec<Row> = Vec::with_capacity(taken.len());
for base in taken {
next.extend(probe_rows(ctx, index, t, base));
}
rows = next;
} else {
let Some((rel, rel_slots)) = pattern_rows(ctx, index, &t) else {
return Vec::new();
};
rows = hash_join(rows, &bound, rel, &rel_slots);
}
for s in pattern_slots(&t) {
if !bound.contains(&s) {
bound.push(s);
}
}
if rows.is_empty() {
break;
}
}
rows
}
fn try_merge_join(
ctx: &Ctx,
index: &GraphIndex,
ta: &(SlotTerm, SlotTerm, SlotTerm),
tb: &(SlotTerm, SlotTerm, SlotTerm),
) -> Option<(Vec<Row>, Vec<usize>)> {
let dict = ctx.rete.dictionary();
let sa = pattern_slots(ta);
let sb = pattern_slots(tb);
let shared: Vec<usize> = sa.iter().copied().filter(|s| sb.contains(s)).collect();
if shared.len() != 1 {
return None;
}
let v = shared[0];
let col_of = |t: &(SlotTerm, SlotTerm, SlotTerm)| -> Option<usize> {
[&t.0, &t.1, &t.2]
.iter()
.position(|x| matches!(x, SlotTerm::Var(i) if *i == v))
};
let ca = col_of(ta)?;
let cb = col_of(tb)?;
if ca != cb {
return None; }
let (lo, hi) = {
let (a, b) = (pattern_scan_bytes(index, ta), pattern_scan_bytes(index, tb));
(a.min(b), a.max(b))
};
if hi >= FAT_SCAN_BYTES && hi / 4 >= lo {
return None;
}
let lower = |t: &(SlotTerm, SlotTerm, SlotTerm)| -> Option<Pattern> {
Some((
const_subject(&t.0, dict)?,
const_predicate(&t.1)?,
const_object(&t.2, dict)?,
))
};
let collect = |t: &(SlotTerm, SlotTerm, SlotTerm), col: usize| -> Option<Vec<(u32, Row)>> {
let pat = lower(t)?;
let mut out = Vec::new();
for tri in index.scan_iter_sorted_on(pat, col)? {
if let Some(r) = triple_row(ctx, t, tri) {
out.push(([tri.0, tri.1, tri.2][col], r));
}
}
Some(out)
};
let rows_a = collect(ta, ca)?;
let rows_b = collect(tb, cb)?;
let mut out: Vec<Row> = Vec::new();
let (mut i, mut j) = (0usize, 0usize);
while i < rows_a.len() && j < rows_b.len() {
match rows_a[i].0.cmp(&rows_b[j].0) {
std::cmp::Ordering::Less => i += 1,
std::cmp::Ordering::Greater => j += 1,
std::cmp::Ordering::Equal => {
let key = rows_a[i].0;
let (i0, j0) = (i, j);
while i < rows_a.len() && rows_a[i].0 == key {
i += 1;
}
while j < rows_b.len() && rows_b[j].0 == key {
j += 1;
}
for (_, ra) in &rows_a[i0..i] {
for (_, rb) in &rows_b[j0..j] {
let mut row = ra.clone();
let mut ok = true;
for (slot, val) in rb.iter().enumerate() {
if let Some(val) = val {
match &row[slot] {
Some(existing) if existing != val => {
ok = false;
break;
}
_ => row[slot] = Some(val.clone()),
}
}
}
if ok {
out.push(row);
}
}
}
}
}
}
let mut slots = pattern_slots(ta);
for s in pattern_slots(tb) {
if !slots.contains(&s) {
slots.push(s);
}
}
Some((out, slots))
}
const BGP_PROBE_THRESHOLD: usize = 512;
const REMOTE_PROBE_MIN: usize = 8;
const REMOTE_PROBE_MAX: usize = 1024;
fn remote_probe_max(index: &GraphIndex) -> usize {
REMOTE_PROBE_MAX * index.read_concurrency().clamp(1, 16)
}
pub(crate) fn prefetch_plan_probes(ctx: &Ctx, index: &GraphIndex, plan: &ProbePlan, rows: &[Row]) {
if !index.is_remote() {
return;
}
let Some(t) = plan.pats.first() else {
return;
};
let mut pats: Vec<Pattern> = Vec::new();
for base in rows {
if let (Some(s), Some(p), Some(o)) = (
probe_subject(ctx, &t.0, base),
probe_predicate(ctx, &t.1, base),
probe_object(ctx, &t.2, base),
) {
if s.is_some() || o.is_some() {
pats.push((s, p, o));
}
}
}
index.prefetch_probe_tiles(&pats);
}
pub(crate) fn prefetch_subject_probes(
ctx: &Ctx,
index: &GraphIndex,
patterns: &[TriplePattern],
sv: &str,
subject_ids: &[u32],
) {
if !index.is_remote() {
return;
}
let Some(lowered) = lower(patterns, ctx) else {
return;
};
let Some(slot) = ctx.slots.slot(sv) else {
return;
};
let dict = ctx.rete.dictionary();
let mut pats: Vec<Pattern> = Vec::new();
for &sid in subject_ids {
let mut base = ctx.slots.empty_row();
base[slot] = Some(Val::Id(dict.subject_node(sid) as i64));
for t in &lowered {
if !matches!(t.0, SlotTerm::Var(i) if i == slot) {
continue;
}
if let (Some(s), Some(p), Some(o)) = (
probe_subject(ctx, &t.0, &base),
probe_predicate(ctx, &t.1, &base),
probe_object(ctx, &t.2, &base),
) {
pats.push((s, p, o));
}
}
}
index.prefetch_probe_tiles(&pats);
}
fn pattern_is_selective(t: &(SlotTerm, SlotTerm, SlotTerm)) -> bool {
matches!(t.0, SlotTerm::Node(_)) || matches!(t.2, SlotTerm::Node(_))
}
fn pattern_slots(t: &(SlotTerm, SlotTerm, SlotTerm)) -> Vec<usize> {
let mut slots: Vec<usize> = Vec::new();
for term in [&t.0, &t.1, &t.2] {
if let SlotTerm::Var(i) = term {
if !slots.contains(i) {
slots.push(*i);
}
}
}
slots
}
fn triple_row(
ctx: &Ctx,
t: &(SlotTerm, SlotTerm, SlotTerm),
(s_id, p_id, o_id): (u32, u32, u32),
) -> Option<Row> {
let dict = ctx.rete.dictionary();
let s_val = dict.subject_node(s_id) as i64;
let p_val = ctx.resolver.canon_id(pred_tag(p_id));
let o_val = dict.object_node(o_id) as i64;
let mut row = ctx.slots.empty_row();
for (term, val) in [(&t.0, s_val), (&t.1, p_val), (&t.2, o_val)] {
if let SlotTerm::Var(i) = term {
match row[*i] {
Some(Val::Id(existing)) if existing != val => return None,
Some(_) => {}
None => row[*i] = Some(Val::Id(val)),
}
}
}
Some(row)
}
fn scan_rows<'q>(
ctx: &'q Ctx<'q>,
index: &'q GraphIndex,
t: (SlotTerm, SlotTerm, SlotTerm),
) -> Option<impl Iterator<Item = Row> + 'q> {
let dict = ctx.rete.dictionary();
let (sid, pid, oid) = (
const_subject(&t.0, dict)?,
const_predicate(&t.1)?,
const_object(&t.2, dict)?,
);
Some(
index
.scan_iter((sid, pid, oid))
.filter_map(move |triple| triple_row(ctx, &t, triple)),
)
}
fn pattern_rows(
ctx: &Ctx,
index: &GraphIndex,
t: &(SlotTerm, SlotTerm, SlotTerm),
) -> Option<(Vec<Row>, Vec<usize>)> {
let slots = pattern_slots(t);
let mut rel: Vec<Row> = Vec::new();
let dict = ctx.rete.dictionary();
let (sid, pid, oid) = (
const_subject(&t.0, dict)?,
const_predicate(&t.1)?,
const_object(&t.2, dict)?,
);
for triple in index.scan_iter((sid, pid, oid)) {
if let Some(row) = triple_row(ctx, t, triple) {
rel.push(row);
}
}
Some((rel, slots))
}
fn pattern_estimates(
ctx: &Ctx,
lowered: &[(SlotTerm, SlotTerm, SlotTerm)],
seed: &std::collections::HashSet<usize>,
) -> Option<Vec<f64>> {
let pyr = ctx.rete.pyramid_if_loaded()?;
let mut pred: std::collections::HashMap<u32, u64> = std::collections::HashMap::new();
for e in &pyr.summary {
*pred.entry(e.predicate).or_insert(0) += e.count as u64;
}
let stats: std::collections::HashMap<u32, &crate::meta::PredStat> = pyr
.predicate_stats
.iter()
.map(|s| (s.predicate, s))
.collect();
let total = ctx.rete.header().quad_count.max(1) as f64;
let num_preds = pred.len().max(1) as f64;
const SEL_SUBJECT: f64 = 0.001;
const SEL_OBJECT: f64 = 0.02;
let node_bound = |t: &SlotTerm| match t {
SlotTerm::Node(_) => true,
SlotTerm::Var(v) => seed.contains(v),
SlotTerm::Pred(_) => false,
};
Some(
lowered
.iter()
.map(|t| {
let (base, st) = match t.1 {
SlotTerm::Pred(p) => {
(*pred.get(&p).unwrap_or(&0) as f64, stats.get(&p).copied())
}
SlotTerm::Var(v) if seed.contains(&v) => (total / num_preds, None),
_ => (total, None),
};
let mut est = base.max(1.0);
if node_bound(&t.0) {
est *= match st {
Some(s) if s.distinct_subjects > 0 => 1.0 / s.distinct_subjects as f64,
_ => SEL_SUBJECT,
};
}
if node_bound(&t.2) {
est *= match st {
Some(s) if s.distinct_objects > 0 => 1.0 / s.distinct_objects as f64,
_ => SEL_OBJECT,
};
}
est.max(1.0)
})
.collect(),
)
}
pub(crate) fn bgp_min_scan_bytes(
ctx: &Ctx,
index: &GraphIndex,
patterns: &[TriplePattern],
) -> Option<u64> {
let lowered = lower(patterns, ctx)?;
lowered.iter().map(|t| pattern_scan_bytes(index, t)).min()
}
fn pattern_scan_bytes(index: &GraphIndex, t: &(SlotTerm, SlotTerm, SlotTerm)) -> u64 {
let sections = index.tile_sections();
let comp = |role: usize| match role {
0 => &t.0,
1 => &t.1,
_ => &t.2,
};
let mut best = u64::MAX;
for perm in crate::index::ALL_PERMS {
let tiles = sections[perm.section_index()];
best = best.min(match comp(perm.roles()[0]) {
SlotTerm::Node(id) | SlotTerm::Pred(id) => tiles
.iter()
.filter(|tile| {
let (lo, hi) = tile.leading_range();
lo <= *id && *id <= hi
})
.map(Tile::encoded_len)
.sum(),
SlotTerm::Var(_) => tiles.iter().map(Tile::encoded_len).sum(),
});
}
best
}
pub(crate) const FAT_SCAN_BYTES: u64 = 2 << 20;
fn selectivity_order(ctx: &Ctx, lowered: &[(SlotTerm, SlotTerm, SlotTerm)]) -> Vec<usize> {
selectivity_order_seeded(ctx, lowered, &std::collections::HashSet::new())
}
fn selectivity_order_seeded(
ctx: &Ctx,
lowered: &[(SlotTerm, SlotTerm, SlotTerm)],
seed: &std::collections::HashSet<usize>,
) -> Vec<usize> {
let estimates = pattern_estimates(ctx, lowered, seed);
let consts = |t: &(SlotTerm, SlotTerm, SlotTerm)| {
[&t.0, &t.1, &t.2]
.into_iter()
.filter(|x| match x {
SlotTerm::Var(v) => seed.contains(v),
_ => true,
})
.count()
};
let vars = |t: &(SlotTerm, SlotTerm, SlotTerm)| -> Vec<usize> {
[&t.0, &t.1, &t.2]
.into_iter()
.filter_map(|x| match x {
SlotTerm::Var(v) => Some(*v),
_ => None,
})
.collect()
};
let score = |i: usize| -> f64 {
match &estimates {
Some(e) => -e[i],
None => consts(&lowered[i]) as f64,
}
};
let type_pid = ctx.rete.dictionary().predicate_id(crate::file::RDF_TYPE);
let is_class_enum = |i: usize, bound: &std::collections::HashSet<usize>| -> bool {
let t = &lowered[i];
type_pid.is_some_and(|tp| matches!(t.1, SlotTerm::Pred(p) if p == tp))
&& matches!(t.2, SlotTerm::Node(_))
&& matches!(t.0, SlotTerm::Var(v) if !bound.contains(&v))
};
let n = lowered.len();
let mut remaining: Vec<usize> = (0..n).collect();
let mut order: Vec<usize> = Vec::with_capacity(n);
let mut bound: std::collections::HashSet<usize> = seed.clone();
while !remaining.is_empty() {
let best = *remaining
.iter()
.max_by(|&&a, &&b| {
let connected = |i: usize| vars(&lowered[i]).iter().any(|v| bound.contains(v));
connected(a)
.cmp(&connected(b))
.then_with(|| is_class_enum(b, &bound).cmp(&is_class_enum(a, &bound)))
.then_with(|| {
score(a)
.partial_cmp(&score(b))
.unwrap_or(std::cmp::Ordering::Equal)
})
.then_with(|| b.cmp(&a))
})
.unwrap();
for v in vars(&lowered[best]) {
bound.insert(v);
}
order.push(best);
remaining.retain(|&i| i != best);
}
order
}
pub(crate) fn bgp_exists(ctx: &Ctx, index: &GraphIndex, patterns: &[TriplePattern]) -> bool {
let dict = ctx.rete.dictionary();
let Some(lowered) = lower(patterns, ctx) else {
return false;
};
if let [t] = lowered.as_slice() {
let names: Vec<usize> = [&t.0, &t.1, &t.2]
.into_iter()
.filter_map(|x| match x {
SlotTerm::Var(v) => Some(*v),
_ => None,
})
.collect();
let distinct = names
.iter()
.enumerate()
.all(|(i, v)| !names[i + 1..].contains(v));
if distinct {
return match (
const_subject(&t.0, dict),
const_predicate(&t.1),
const_object(&t.2, dict),
) {
(Some(s), Some(p), Some(o)) => index.scan_iter((s, p, o)).next().is_some(),
_ => false,
};
}
}
!eval_bgp_rows(ctx, index, patterns).is_empty()
}
pub(crate) struct BgpSolutions<'q> {
scan: Option<Box<dyn Iterator<Item = Row> + 'q>>,
prefix: Vec<Row>,
buckets: HashMap<Vec<Val>, Vec<usize>>,
shared: Vec<usize>,
cartesian: bool,
seed_only: bool,
cur_scan: Option<Row>,
matches: Vec<usize>,
mi: usize,
}
impl<'q> BgpSolutions<'q> {
fn empty() -> Self {
BgpSolutions {
scan: None,
prefix: Vec::new(),
buckets: HashMap::new(),
shared: Vec::new(),
cartesian: false,
seed_only: false,
cur_scan: None,
matches: Vec::new(),
mi: 0,
}
}
pub(crate) fn new(ctx: &'q Ctx<'q>, index: &'q GraphIndex, patterns: &[TriplePattern]) -> Self {
if patterns.is_empty() {
return BgpSolutions {
scan: Some(Box::new(std::iter::once(ctx.slots.empty_row()))),
prefix: vec![ctx.slots.empty_row()],
buckets: HashMap::new(),
shared: Vec::new(),
cartesian: true,
seed_only: true,
cur_scan: None,
matches: Vec::new(),
mi: 0,
};
}
let Some(lowered) = lower(patterns, ctx) else {
return Self::empty();
};
let order = selectivity_order(ctx, &lowered);
let (&last_i, prefix_is) = order.split_last().unwrap();
let prefix_pats: Vec<TriplePattern> =
prefix_is.iter().map(|&i| patterns[i].clone()).collect();
let prefix = eval_bgp_rows(ctx, index, &prefix_pats);
if prefix.is_empty() {
return Self::empty();
}
let Some(scan) = scan_rows(ctx, index, lowered[last_i]) else {
return Self::empty();
};
let shared: Vec<usize> = pattern_slots(&lowered[last_i])
.into_iter()
.filter(|&s| prefix[0][s].is_some())
.collect();
let cartesian = shared.is_empty();
let seed_only = prefix.len() == 1 && prefix[0].iter().all(Option::is_none);
let mut buckets: HashMap<Vec<Val>, Vec<usize>> = HashMap::new();
if !cartesian {
for (i, r) in prefix.iter().enumerate() {
let key: Vec<Val> = shared.iter().map(|&s| r[s].clone().unwrap()).collect();
buckets.entry(key).or_default().push(i);
}
}
BgpSolutions {
scan: Some(Box::new(scan)),
prefix,
buckets,
shared,
cartesian,
seed_only,
cur_scan: None,
matches: Vec::new(),
mi: 0,
}
}
}
impl Iterator for BgpSolutions<'_> {
type Item = Row;
fn next(&mut self) -> Option<Row> {
if self.seed_only {
return self.scan.as_mut()?.next();
}
loop {
if self.mi < self.matches.len() {
let pi = self.matches[self.mi];
self.mi += 1;
let mut merged = self.prefix[pi].clone();
for (slot, v) in self.cur_scan.as_ref().unwrap().iter().enumerate() {
if v.is_some() {
merged[slot] = v.clone();
}
}
return Some(merged);
}
let s = self.scan.as_mut()?.next()?;
self.matches.clear();
if self.cartesian {
self.matches.extend(0..self.prefix.len());
} else {
let key: Vec<Val> = self.shared.iter().map(|&i| s[i].clone().unwrap()).collect();
if let Some(c) = self.buckets.get(&key) {
self.matches.extend_from_slice(c);
}
}
self.mi = 0;
self.cur_scan = Some(s);
}
}
}
fn probe_subject(ctx: &Ctx, t: &SlotTerm, base: &Row) -> Option<Option<u32>> {
let dict = ctx.rete.dictionary();
match t {
SlotTerm::Node(n) => dict.node_as_subject_id(*n).map(Some),
SlotTerm::Pred(_) => None,
SlotTerm::Var(i) => match &base[*i] {
None => Some(None),
Some(Val::Id(v)) if *v >= 0 => dict.node_as_subject_id(*v as u32).map(Some),
Some(_) => None,
},
}
}
fn probe_object(ctx: &Ctx, t: &SlotTerm, base: &Row) -> Option<Option<u32>> {
let dict = ctx.rete.dictionary();
match t {
SlotTerm::Node(n) => dict.node_as_object_id(*n).map(Some),
SlotTerm::Pred(_) => None,
SlotTerm::Var(i) => match &base[*i] {
None => Some(None),
Some(Val::Id(v)) if *v >= 0 => dict.node_as_object_id(*v as u32).map(Some),
Some(_) => None,
},
}
}
fn probe_predicate(ctx: &Ctx, t: &SlotTerm, base: &Row) -> Option<Option<u32>> {
let dict = ctx.rete.dictionary();
match t {
SlotTerm::Pred(p) => Some(Some(*p)),
SlotTerm::Node(_) => None,
SlotTerm::Var(i) => match &base[*i] {
None => Some(None),
Some(Val::Id(v)) if *v < 0 => Some(Some((-v - 1) as u32)),
Some(Val::Id(v)) => ctx
.resolver
.term(*v)
.and_then(|t| dict.predicate_id(&t))
.map(Some),
Some(Val::Str(_)) => None,
},
}
}
fn probe_rows<'q>(
ctx: &'q Ctx<'q>,
index: &'q GraphIndex,
t: (SlotTerm, SlotTerm, SlotTerm),
base: Row,
) -> Box<dyn Iterator<Item = Row> + 'q> {
let (Some(sid), Some(pid), Some(oid)) = (
probe_subject(ctx, &t.0, &base),
probe_predicate(ctx, &t.1, &base),
probe_object(ctx, &t.2, &base),
) else {
return Box::new(std::iter::empty());
};
let dict = ctx.rete.dictionary();
Box::new(
index
.scan_iter((sid, pid, oid))
.filter_map(move |(s_id, p_id, o_id)| {
let s_val = dict.subject_node(s_id) as i64;
let p_val = ctx.resolver.canon_id(pred_tag(p_id));
let o_val = dict.object_node(o_id) as i64;
let mut row = base.clone();
for (term, val) in [(&t.0, s_val), (&t.1, p_val), (&t.2, o_val)] {
if let SlotTerm::Var(i) = term {
match row[*i] {
Some(Val::Id(existing)) if existing != val => return None,
Some(Val::Id(_)) => {}
Some(Val::Str(_)) => return None,
None => row[*i] = Some(Val::Id(val)),
}
}
}
Some(row)
}),
)
}
pub(crate) struct ProbePlan {
pats: Vec<(SlotTerm, SlotTerm, SlotTerm)>,
}
impl ProbePlan {
pub(crate) fn new(ctx: &Ctx, patterns: &[TriplePattern], seed_mask: &[bool]) -> Option<Self> {
let lowered = lower(patterns, ctx)?;
let seed: std::collections::HashSet<usize> = seed_mask
.iter()
.enumerate()
.filter_map(|(i, b)| b.then_some(i))
.collect();
let order = selectivity_order_seeded(ctx, &lowered, &seed);
Some(ProbePlan {
pats: order.into_iter().map(|i| lowered[i]).collect(),
})
}
}
pub(crate) struct ProbeJoin<'q> {
ctx: &'q Ctx<'q>,
index: &'q GraphIndex,
pats: Vec<(SlotTerm, SlotTerm, SlotTerm)>,
stack: Vec<Box<dyn Iterator<Item = Row> + 'q>>,
}
impl<'q> ProbeJoin<'q> {
pub(crate) fn new(
ctx: &'q Ctx<'q>,
index: &'q GraphIndex,
patterns: &[TriplePattern],
) -> Option<Self> {
let plan = ProbePlan::new(ctx, patterns, &vec![false; ctx.slots.len()])?;
Some(Self::from_plan(ctx, index, &plan, ctx.slots.empty_row()))
}
pub(crate) fn from_plan(
ctx: &'q Ctx<'q>,
index: &'q GraphIndex,
plan: &ProbePlan,
seed: Row,
) -> Self {
let pats = plan.pats.clone();
let first = probe_rows(ctx, index, pats[0], seed);
ProbeJoin {
ctx,
index,
pats,
stack: vec![first],
}
}
}
impl Iterator for ProbeJoin<'_> {
type Item = Row;
fn next(&mut self) -> Option<Row> {
loop {
let depth = self.stack.len();
match self.stack.last_mut()?.next() {
Some(row) => {
if depth == self.pats.len() {
return Some(row);
}
let it = probe_rows(self.ctx, self.index, self.pats[depth], row);
self.stack.push(it);
}
None => {
self.stack.pop();
if self.stack.is_empty() {
return None;
}
}
}
}
}
}
fn hash_join(
left: Vec<Row>,
left_bound: &[usize],
right: Vec<Row>,
right_slots: &[usize],
) -> Vec<Row> {
if left.len() == 1 && left_bound.is_empty() {
return right;
}
if left.is_empty() || right.is_empty() {
return Vec::new();
}
let shared: Vec<usize> = right_slots
.iter()
.copied()
.filter(|s| left_bound.contains(s))
.collect();
let fill = |l: &Row, r: &Row| -> Row {
let mut out = l.clone();
for &s in right_slots {
out[s] = r[s].clone();
}
out
};
if shared.is_empty() {
let mut out = Vec::with_capacity(left.len() * right.len());
for l in &left {
for r in &right {
out.push(fill(l, r));
}
}
return out;
}
let key_of = |b: &Row| -> Vec<Val> { shared.iter().map(|&s| b[s].clone().unwrap()).collect() };
let mut out = Vec::new();
if right.len() <= left.len() {
let mut buckets: HashMap<Vec<Val>, Vec<Row>> = HashMap::new();
for r in right {
buckets.entry(key_of(&r)).or_default().push(r);
}
for l in &left {
if let Some(rs) = buckets.get(&key_of(l)) {
for r in rs {
out.push(fill(l, r));
}
}
}
} else {
let mut buckets: HashMap<Vec<Val>, Vec<Row>> = HashMap::new();
for l in left {
buckets.entry(key_of(&l)).or_default().push(l);
}
for r in &right {
if let Some(ls) = buckets.get(&key_of(r)) {
for l in ls {
let mut row = r.clone();
for &s in left_bound {
if let Some(v) = &l[s] {
row[s] = Some(v.clone());
}
}
out.push(row);
}
}
}
}
out
}
fn const_subject(t: &SlotTerm, d: &crate::Dictionary) -> Option<Option<u32>> {
match t {
SlotTerm::Node(n) => d.node_as_subject_id(*n).map(Some),
SlotTerm::Pred(_) => None, SlotTerm::Var(_) => Some(None),
}
}
fn const_object(t: &SlotTerm, d: &crate::Dictionary) -> Option<Option<u32>> {
match t {
SlotTerm::Node(n) => d.node_as_object_id(*n).map(Some),
SlotTerm::Pred(_) => None,
SlotTerm::Var(_) => Some(None),
}
}
fn const_predicate(t: &SlotTerm) -> Option<Option<u32>> {
match t {
SlotTerm::Pred(p) => Some(Some(*p)),
SlotTerm::Node(_) => None, SlotTerm::Var(_) => Some(None),
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::dictionary::DictionaryBuilder;
use crate::file::write_file;
use crate::index::GraphIndexBuilder;
fn rete_from(triples: &[(&str, &str, &str)]) -> Vec<u8> {
let mut db = DictionaryBuilder::new();
for (s, p, o) in triples {
db.observe(s, p, o);
}
let dict = db.build();
let mut ib = GraphIndexBuilder::new();
for (s, p, o) in triples {
ib.push(dict.encode(s, p, o).unwrap());
}
write_file(&dict, &ib.build(), false, &[], 0)
}
fn pat(s: &str, p: &str, o: &str) -> TriplePattern {
TriplePattern {
s: PatternTerm::parse(s),
p: PatternTerm::parse(p),
o: PatternTerm::parse(o),
}
}
#[test]
fn single_pattern_binds_variable() {
let bytes = rete_from(&[("Alice", "knows", "Bob"), ("Bob", "knows", "Carol")]);
let rete = Rete::open(&bytes).unwrap();
let sols = eval_bgp(&rete, &[pat("Alice", "knows", "?y")]);
assert_eq!(sols.len(), 1);
assert_eq!(sols[0]["y"], "Bob");
}
#[test]
fn two_hop_join_on_shared_variable() {
let bytes = rete_from(&[
("Alice", "knows", "Bob"),
("Bob", "knows", "Carol"),
("Carol", "knows", "Dave"),
("Alice", "knows", "Eve"), ]);
let rete = Rete::open(&bytes).unwrap();
let sols = eval_bgp(&rete, &[pat("?x", "knows", "?y"), pat("?y", "knows", "?z")]);
let mut got: Vec<_> = sols
.iter()
.map(|b| (b["x"].clone(), b["y"].clone(), b["z"].clone()))
.collect();
got.sort();
assert_eq!(
got,
vec![
("Alice".into(), "Bob".into(), "Carol".into()),
("Bob".into(), "Carol".into(), "Dave".into()),
]
);
}
#[test]
fn repeated_variable_within_pattern() {
let bytes = rete_from(&[("Alice", "knows", "Bob"), ("Bob", "knows", "Bob")]);
let rete = Rete::open(&bytes).unwrap();
let sols = eval_bgp(&rete, &[pat("?x", "knows", "?x")]);
assert_eq!(sols.len(), 1);
assert_eq!(sols[0]["x"], "Bob");
}
#[test]
fn no_solutions_yields_empty() {
let bytes = rete_from(&[("Alice", "knows", "Bob")]);
let rete = Rete::open(&bytes).unwrap();
assert!(eval_bgp(&rete, &[pat("Alice", "likes", "?y")]).is_empty());
}
#[test]
fn snowflake_type_and_adjacency_matches_reference() {
const TYPE: &str = "<http://www.w3.org/1999/02/22-rdf-syntax-ns#type>";
let mut triples: Vec<(String, String, String)> = Vec::new();
for i in 0..60 {
let s = format!("S{i}");
triples.push((s.clone(), TYPE.into(), "Student".into()));
triples.push((s.clone(), "takesCourse".into(), format!("c{}", i % 5)));
triples.push((s, "takesCourse".into(), format!("c{}", (i + 1) % 5)));
}
for j in 0..5 {
triples.push((format!("c{j}"), TYPE.into(), "Course".into()));
}
triples.push(("Prof".into(), "teacherOf".into(), "c0".into()));
triples.push(("Prof".into(), "teacherOf".into(), "c2".into()));
let refs: Vec<(&str, &str, &str)> = triples
.iter()
.map(|(s, p, o)| (s.as_str(), p.as_str(), o.as_str()))
.collect();
let bytes = rete_from(&refs);
let rete = Rete::open(&bytes).unwrap();
let got: std::collections::BTreeSet<(String, String)> = eval_bgp(
&rete,
&[
pat("?x", TYPE, "Student"),
pat("?y", TYPE, "Course"),
pat("?x", "takesCourse", "?y"),
pat("Prof", "teacherOf", "?y"),
],
)
.into_iter()
.map(|b| (b["x"].clone(), b["y"].clone()))
.collect();
let taught = ["c0", "c2"];
let mut want = std::collections::BTreeSet::new();
for i in 0..60 {
for c in [i % 5, (i + 1) % 5] {
let course = format!("c{c}");
if taught.contains(&course.as_str()) {
want.insert((format!("S{i}"), course));
}
}
}
assert_eq!(got, want, "snowflake join must match the brute-force set");
assert!(!want.is_empty(), "sanity: the reference set is non-empty");
}
#[test]
fn merge_join_subject_star_cross_product() {
let bytes = rete_from(&[
("x", "a", "a1"),
("x", "a", "a2"),
("x", "b", "b1"),
("x", "b", "b2"),
("y", "a", "a3"),
("y", "b", "b3"),
("z", "a", "a4"), ]);
let rete = Rete::open(&bytes).unwrap();
let mut got: Vec<(String, String, String)> =
eval_bgp(&rete, &[pat("?x", "a", "?av"), pat("?x", "b", "?bv")])
.iter()
.map(|m| (m["x"].clone(), m["av"].clone(), m["bv"].clone()))
.collect();
got.sort();
let mut want: Vec<(String, String, String)> = [
("x", "a1", "b1"),
("x", "a1", "b2"),
("x", "a2", "b1"),
("x", "a2", "b2"),
("y", "a3", "b3"),
]
.into_iter()
.map(|(a, b, c)| (a.to_string(), b.to_string(), c.to_string()))
.collect();
want.sort();
assert_eq!(got, want, "same-role merge must equal the brute-force join");
}
}