Skip to main content

rete_core/
bgp.rs

1//! Basic Graph Pattern (BGP) evaluation — the core of SPARQL (SPEC.md §8,
2//! stage 1). A BGP is a set of triple patterns whose variables join on equality;
3//! evaluating it yields variable bindings.
4//!
5//! Evaluation is a left-deep **hash join**: each pattern is scanned from the
6//! index *once* (binding only its own constant terms), producing a relation of
7//! candidate rows; that relation is then joined against the running solution set
8//! on the variables they share. This is O(scan + matches) per pattern rather
9//! than the O(bindings × scan) of a per-binding nested-loop probe — the
10//! difference between sub-second and minutes once a pattern binds tens of
11//! thousands of rows. Correctness does not depend on pattern order.
12//!
13//! Solutions are slot `Row`s of tagged dictionary ids (see `crate::row`):
14//! joins hash and compare integers, and terms are resolved to strings only at
15//! the engine's projection boundary, never per intermediate row.
16
17use std::collections::{BTreeMap, HashMap};
18
19use crate::file::Rete;
20use crate::index::{GraphIndex, Pattern, Tile};
21use crate::row::{Ctx, Row, Slots, Val};
22
23/// A term in a pattern: a named variable or a constant term token.
24#[derive(Debug, Clone, PartialEq, Eq)]
25pub enum PatternTerm {
26    Var(String),
27    Const(String),
28}
29
30impl PatternTerm {
31    /// `?x` → variable `x`; anything else → a constant.
32    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/// A triple pattern `(subject, predicate, object)` of [`PatternTerm`]s.
42#[derive(Debug, Clone)]
43pub struct TriplePattern {
44    pub s: PatternTerm,
45    pub p: PatternTerm,
46    pub o: PatternTerm,
47}
48
49/// A solution: variable name → bound term (the public, resolved form).
50pub type Binding = BTreeMap<String, String>;
51
52// --- integer join core ------------------------------------------------------
53//
54// Variables bind to a tagged i64: a node ID `n` is stored as `n` (>= 0), a
55// predicate ID `p` as `-(p+1)` (< 0). Nodes are the unified subject/object space
56// so a variable joins consistently across subject and object positions. A
57// predicate value whose term is also a node is canonicalized to the node id
58// (`Resolver::canon_id`), so cross-role joins match exactly when the term
59// strings match.
60
61/// A pattern position lowered to slot/integer space.
62#[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
73/// Register every variable in `patterns` with the slot map.
74pub(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
84/// Lower patterns to slot/integer space; `None` if a constant term is unknown
85/// (making the whole BGP unsatisfiable) or a variable has no slot.
86fn 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
107/// Evaluate a BGP against the file's default graph, returning all solutions
108/// resolved to terms (the public convenience API).
109pub 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
119/// Resolve a row to a named-term binding (every bound slot). Uses the uncached
120/// decode — this is the output boundary, where terms are typically seen once.
121pub(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
133/// Evaluate a BGP to slot rows against a specific graph `index` (joins run on
134/// integer ids; the shared dictionary comes from `ctx`).
135pub(crate) fn eval_bgp_rows(ctx: &Ctx, index: &GraphIndex, patterns: &[TriplePattern]) -> Vec<Row> {
136    // An empty BGP has exactly one (empty) solution.
137    if patterns.is_empty() {
138        return vec![ctx.slots.empty_row()];
139    }
140    // Lower all patterns; a missing constant term makes the BGP empty.
141    let Some(lowered) = lower(patterns, ctx) else {
142        return Vec::new();
143    };
144
145    // Join the patterns most-constrained-first (and keeping the join connected),
146    // so intermediate relations stay small. Pure reordering: the hash join is
147    // order-independent, so the result is unchanged.
148    let order = selectivity_order(ctx, &lowered);
149    let mut rows: Vec<Row> = vec![ctx.slots.empty_row()];
150    let mut bound: Vec<usize> = Vec::new();
151    let mut merged: [bool; 2] = [false, false];
152
153    // Merge-join seed: if the two cheapest patterns share a same-role variable the
154    // 6-permutation index can co-sort them on, join them with a linear two-pointer
155    // **merge** (no hash table) and skip both below. Falls back silently to the
156    // hash/probe path when the shape doesn't qualify.
157    if order.len() >= 2 {
158        if let Some((rel, slots)) =
159            try_merge_join(ctx, index, &lowered[order[0]], &lowered[order[1]])
160        {
161            rows = rel;
162            bound = slots;
163            merged = [true, true];
164        }
165    }
166
167    for (k, &idx) in order.iter().enumerate() {
168        if k < 2 && merged[k] {
169            continue; // already consumed by the merge seed
170        }
171        let t = lowered[idx];
172        // Hybrid join: once the running result is small and the next pattern shares
173        // an already-bound variable, **probe** it per row through the index instead
174        // of scanning its whole extent and hash-joining. A left-deep hash join
175        // otherwise materializes every pattern's full scan even when only a handful
176        // of rows survive — e.g. a `?x rdf:type C` over thousands of instances when
177        // the prefix already pinned ?x to 26 rows. Probing turns that O(scan) into
178        // O(rows × lookup). The full scan + hash join stays the path when the prefix
179        // is large (one scan beats many probes) or the pattern is a cartesian join
180        // (no shared bound variable to constrain the probe).
181        let shares_bound = pattern_slots(&t).iter().any(|s| bound.contains(s));
182        // Decide probe (per-row index lookup) vs scan + hash join. In memory a
183        // probe is a cheap lookup, so we probe up to a moderate prefix. Remotely
184        // each probe is a byte-range round-trip: probing a 500-row prefix is 500
185        // sequential fetches, while scanning a selectively-bound pattern is a few
186        // coalesced reads — so remotely we only probe a broad (whole-predicate)
187        // pattern, or a tiny prefix, and never a large one.
188        let do_probe = !bound.is_empty()
189            && shares_bound
190            && !rows.is_empty()
191            && if index.is_remote() {
192                rows.len() <= remote_probe_max(index)
193                    && (rows.len() <= REMOTE_PROBE_MIN || !pattern_is_selective(&t))
194            } else {
195                rows.len() <= BGP_PROBE_THRESHOLD
196            };
197        if do_probe {
198            let taken = std::mem::take(&mut rows);
199            // Over a lazy reader each probe is a byte-range round trip, and they run
200            // one at a time. First fault the tiles the *whole* batch will route to,
201            // together — one coalesced parallel read per section instead of N
202            // sequential ones — so the probes below hit a warm tile cache. No-op
203            // for a local index (every tile is already resident).
204            if index.is_remote() {
205                let pats: Vec<Pattern> = taken
206                    .iter()
207                    .filter_map(|base| {
208                        match (
209                            probe_subject(ctx, &t.0, base),
210                            probe_predicate(ctx, &t.1, base),
211                            probe_object(ctx, &t.2, base),
212                        ) {
213                            (Some(s), Some(p), Some(o)) => Some((s, p, o)),
214                            _ => None,
215                        }
216                    })
217                    .collect();
218                index.prefetch_probe_tiles(&pats);
219            }
220            let mut next: Vec<Row> = Vec::with_capacity(taken.len());
221            for base in taken {
222                next.extend(probe_rows(ctx, index, t, base));
223            }
224            rows = next;
225        } else {
226            let Some((rel, rel_slots)) = pattern_rows(ctx, index, &t) else {
227                return Vec::new();
228            };
229            rows = hash_join(rows, &bound, rel, &rel_slots);
230        }
231        for s in pattern_slots(&t) {
232            if !bound.contains(&s) {
233                bound.push(s);
234            }
235        }
236        if rows.is_empty() {
237            break;
238        }
239    }
240    rows
241}
242
243/// Merge-join two base patterns that share **exactly one** variable sitting in the
244/// **same canonical position** (both subject, or both object) in each. That is the
245/// only shape rete's role-aware id spaces let a sort-merge align: the index sorts
246/// subject-ids and object-ids in independent spaces, so a cross-role (subject↔
247/// object) join's two streams are not co-sorted on the join value. Each side is
248/// streamed from the 6-permutation index already sorted on that column
249/// ([`GraphIndex::scan_iter_sorted_on`]), so the join is a linear two-pointer
250/// merge with no hash table — equal-key runs cross-product. Returns
251/// `(rows, bound_slots)`, or `None` when the shape/index can't support a merge and
252/// the caller should hash-join instead. The result multiset equals the hash
253/// join's (a join is order-independent).
254fn try_merge_join(
255    ctx: &Ctx,
256    index: &GraphIndex,
257    ta: &(SlotTerm, SlotTerm, SlotTerm),
258    tb: &(SlotTerm, SlotTerm, SlotTerm),
259) -> Option<(Vec<Row>, Vec<usize>)> {
260    let dict = ctx.rete.dictionary();
261    let sa = pattern_slots(ta);
262    let sb = pattern_slots(tb);
263    let shared: Vec<usize> = sa.iter().copied().filter(|s| sb.contains(s)).collect();
264    if shared.len() != 1 {
265        return None;
266    }
267    let v = shared[0];
268    let col_of = |t: &(SlotTerm, SlotTerm, SlotTerm)| -> Option<usize> {
269        [&t.0, &t.1, &t.2]
270            .iter()
271            .position(|x| matches!(x, SlotTerm::Var(i) if *i == v))
272    };
273    let ca = col_of(ta)?;
274    let cb = col_of(tb)?;
275    if ca != cb {
276        return None; // cross-role: the two role id-spaces aren't co-sorted
277    }
278    // A merge MATERIALIZES both sides in full (`collect` below builds a Vec per
279    // side). When one side is a pinpoint (a bound leading key routing to a few
280    // KB of tiles) and the other is fat (a broad predicate spanning megabytes),
281    // the linear merge costs building — and, on a remote index, FETCHING — the
282    // fat extent end-to-end: gigabytes over HTTP, and the 32-bit wasm OOM
283    // behind `?x wdt:P279 <C> . ?x rdfs:label ?l` on a 185M-triple remote
284    // graph. The hybrid loop handles that shape strictly better (scan the
285    // pinpoint side, probe the fat one per surviving row), so leave the merge
286    // seed to comparably-sized sides.
287    let (lo, hi) = {
288        let (a, b) = (pattern_scan_bytes(index, ta), pattern_scan_bytes(index, tb));
289        (a.min(b), a.max(b))
290    };
291    if hi >= FAT_SCAN_BYTES && hi / 4 >= lo {
292        return None;
293    }
294    let lower = |t: &(SlotTerm, SlotTerm, SlotTerm)| -> Option<Pattern> {
295        Some((
296            const_subject(&t.0, dict)?,
297            const_predicate(&t.1)?,
298            const_object(&t.2, dict)?,
299        ))
300    };
301    // Materialize each side as `(join-key, row)`, already ascending in the join
302    // column (the index streams it sorted there).
303    let collect = |t: &(SlotTerm, SlotTerm, SlotTerm), col: usize| -> Option<Vec<(u32, Row)>> {
304        let pat = lower(t)?;
305        let mut out = Vec::new();
306        for tri in index.scan_iter_sorted_on(pat, col)? {
307            if let Some(r) = triple_row(ctx, t, tri) {
308                out.push(([tri.0, tri.1, tri.2][col], r));
309            }
310        }
311        Some(out)
312    };
313    let rows_a = collect(ta, ca)?;
314    let rows_b = collect(tb, cb)?;
315
316    let mut out: Vec<Row> = Vec::new();
317    let (mut i, mut j) = (0usize, 0usize);
318    while i < rows_a.len() && j < rows_b.len() {
319        match rows_a[i].0.cmp(&rows_b[j].0) {
320            std::cmp::Ordering::Less => i += 1,
321            std::cmp::Ordering::Greater => j += 1,
322            std::cmp::Ordering::Equal => {
323                let key = rows_a[i].0;
324                let (i0, j0) = (i, j);
325                while i < rows_a.len() && rows_a[i].0 == key {
326                    i += 1;
327                }
328                while j < rows_b.len() && rows_b[j].0 == key {
329                    j += 1;
330                }
331                for (_, ra) in &rows_a[i0..i] {
332                    for (_, rb) in &rows_b[j0..j] {
333                        // Combine the two rows; they agree on the shared slot `v`
334                        // (same key ⇒ same node value at the same column).
335                        let mut row = ra.clone();
336                        let mut ok = true;
337                        for (slot, val) in rb.iter().enumerate() {
338                            if let Some(val) = val {
339                                match &row[slot] {
340                                    Some(existing) if existing != val => {
341                                        ok = false;
342                                        break;
343                                    }
344                                    _ => row[slot] = Some(val.clone()),
345                                }
346                            }
347                        }
348                        if ok {
349                            out.push(row);
350                        }
351                    }
352                }
353            }
354        }
355    }
356    let mut slots = pattern_slots(ta);
357    for s in pattern_slots(tb) {
358        if !slots.contains(&s) {
359            slots.push(s);
360        }
361    }
362    Some((out, slots))
363}
364
365/// Probe the join tail (instead of full-scanning) once the running BGP result is
366/// at most this many rows: each remaining pattern that shares a bound variable is
367/// then resolved by `rows × index-lookup` rather than a whole-extent scan. Kept
368/// small so the probe count is always cheap in absolute terms — the dominant win
369/// is the large-scan-after-tiny-prefix case (a `?x a C` over thousands when the
370/// prefix already pinned ?x to a few dozen rows); a *moderately* large prefix
371/// keeps the one-pass scan + hash join, which beats thousands of probes.
372const BGP_PROBE_THRESHOLD: usize = 512;
373
374/// Remote (lazy) probe bounds. Over HTTP byte-range reads, a per-row probe is a
375/// network round-trip, so we probe far more reluctantly than in memory: only when
376/// scanning the next pattern would be a *broad* read (a whole-predicate scan,
377/// [`pattern_is_selective`] is false), or the prefix is tiny enough that a handful
378/// of probes is trivial either way. A pattern that carries its own bound
379/// subject/object routes to a narrow index range — cheaper to scan once and
380/// hash-join than to probe per prefix row. Probing a large prefix is always a
381/// loss remotely, so it is capped.
382const REMOTE_PROBE_MIN: usize = 8;
383const REMOTE_PROBE_MAX: usize = 1024;
384
385/// The remote probe budget, widened by the reader's concurrency: a serial
386/// sync-XHR reader (a phone without the COI pool) pays one full round trip per
387/// probe, so its budget stays at [`REMOTE_PROBE_MAX`]; a reader overlapping
388/// 16 range reads (the CLI's thread pool, the asyncified fetch variant, the
389/// COI fetch-worker pool) amortizes ~16 probes per round trip and gets a
390/// proportionally larger budget before the one-pass scan wins again.
391fn remote_probe_max(index: &GraphIndex) -> usize {
392    REMOTE_PROBE_MAX * index.read_concurrency().clamp(1, 16)
393}
394
395/// Batch-fault the tiles that a [`ProbePlan`]'s FIRST pattern will route to
396/// across a batch of seed rows — one coalesced read per section instead of a
397/// blocking round trip per row. Only the first pattern is prefetchable (later
398/// ones bind from earlier probe results), and only rows whose seed binds the
399/// pattern's subject or object are included: a predicate-only pattern routes
400/// to the predicate's WHOLE span, which for the fat sides this path serves
401/// would prefetch the very extent the probe strategy exists to avoid.
402pub(crate) fn prefetch_plan_probes(ctx: &Ctx, index: &GraphIndex, plan: &ProbePlan, rows: &[Row]) {
403    if !index.is_remote() {
404        return;
405    }
406    let Some(t) = plan.pats.first() else {
407        return;
408    };
409    let mut pats: Vec<Pattern> = Vec::new();
410    for base in rows {
411        if let (Some(s), Some(p), Some(o)) = (
412            probe_subject(ctx, &t.0, base),
413            probe_predicate(ctx, &t.1, base),
414            probe_object(ctx, &t.2, base),
415        ) {
416            if s.is_some() || o.is_some() {
417                pats.push((s, p, o));
418            }
419        }
420    }
421    index.prefetch_probe_tiles(&pats);
422}
423
424/// Batch-fault the tiles that probing `patterns` from `subject_ids` (each
425/// bound onto the subject variable `sv`) will route to — one coalesced read
426/// per section instead of one round trip per candidate. Purely a cache
427/// warmer: correctness is untouched, and a local index is a no-op. Used by
428/// the FILTER-CONTAINS pushdown before it probes its candidate subjects.
429pub(crate) fn prefetch_subject_probes(
430    ctx: &Ctx,
431    index: &GraphIndex,
432    patterns: &[TriplePattern],
433    sv: &str,
434    subject_ids: &[u32],
435) {
436    if !index.is_remote() {
437        return;
438    }
439    let Some(lowered) = lower(patterns, ctx) else {
440        return;
441    };
442    let Some(slot) = ctx.slots.slot(sv) else {
443        return;
444    };
445    let dict = ctx.rete.dictionary();
446    let mut pats: Vec<Pattern> = Vec::new();
447    for &sid in subject_ids {
448        let mut base = ctx.slots.empty_row();
449        base[slot] = Some(Val::Id(dict.subject_node(sid) as i64));
450        for t in &lowered {
451            if !matches!(t.0, SlotTerm::Var(i) if i == slot) {
452                continue;
453            }
454            if let (Some(s), Some(p), Some(o)) = (
455                probe_subject(ctx, &t.0, &base),
456                probe_predicate(ctx, &t.1, &base),
457                probe_object(ctx, &t.2, &base),
458            ) {
459                pats.push((s, p, o));
460            }
461        }
462    }
463    index.prefetch_probe_tiles(&pats);
464}
465
466/// A pattern whose own constant terms (a bound subject or object) already route
467/// it to a narrow index range — a few coalesced range reads to scan in full.
468fn pattern_is_selective(t: &(SlotTerm, SlotTerm, SlotTerm)) -> bool {
469    matches!(t.0, SlotTerm::Node(_)) || matches!(t.2, SlotTerm::Node(_))
470}
471
472/// The (deduped) slots a lowered pattern binds.
473fn pattern_slots(t: &(SlotTerm, SlotTerm, SlotTerm)) -> Vec<usize> {
474    let mut slots: Vec<usize> = Vec::new();
475    for term in [&t.0, &t.1, &t.2] {
476        if let SlotTerm::Var(i) = term {
477            if !slots.contains(i) {
478                slots.push(*i);
479            }
480        }
481    }
482    slots
483}
484
485/// Build a solution row for one scanned triple, enforcing repeated variables
486/// *within* the pattern (e.g. `?x p ?x`). `None` = the triple doesn't satisfy
487/// a repeated variable.
488fn triple_row(
489    ctx: &Ctx,
490    t: &(SlotTerm, SlotTerm, SlotTerm),
491    (s_id, p_id, o_id): (u32, u32, u32),
492) -> Option<Row> {
493    let dict = ctx.rete.dictionary();
494    let s_val = dict.subject_node(s_id) as i64;
495    let p_val = ctx.resolver.canon_id(pred_tag(p_id));
496    let o_val = dict.object_node(o_id) as i64;
497    let mut row = ctx.slots.empty_row();
498    for (term, val) in [(&t.0, s_val), (&t.1, p_val), (&t.2, o_val)] {
499        if let SlotTerm::Var(i) = term {
500            match row[*i] {
501                Some(Val::Id(existing)) if existing != val => return None,
502                Some(_) => {}
503                None => row[*i] = Some(Val::Id(val)),
504            }
505        }
506    }
507    Some(row)
508}
509
510/// Lazily scan one lowered pattern as a stream of solution rows. The scan
511/// constrains only the pattern's constant terms (bound variables are joined in
512/// afterwards, not pushed into the scan). `None` = a constant is unsatisfiable
513/// (unknown to the dictionary or in an impossible role), which empties the BGP.
514fn scan_rows<'q>(
515    ctx: &'q Ctx<'q>,
516    index: &'q GraphIndex,
517    t: (SlotTerm, SlotTerm, SlotTerm),
518) -> Option<impl Iterator<Item = Row> + 'q> {
519    let dict = ctx.rete.dictionary();
520    let (sid, pid, oid) = (
521        const_subject(&t.0, dict)?,
522        const_predicate(&t.1)?,
523        const_object(&t.2, dict)?,
524    );
525    Some(
526        index
527            .scan_iter((sid, pid, oid))
528            .filter_map(move |triple| triple_row(ctx, &t, triple)),
529    )
530}
531
532/// Scan one lowered pattern into a materialized relation, returning the rows
533/// and the slots the pattern binds. `None` = a constant is unsatisfiable.
534fn pattern_rows(
535    ctx: &Ctx,
536    index: &GraphIndex,
537    t: &(SlotTerm, SlotTerm, SlotTerm),
538) -> Option<(Vec<Row>, Vec<usize>)> {
539    let slots = pattern_slots(t);
540    let mut rel: Vec<Row> = Vec::new();
541    let dict = ctx.rete.dictionary();
542    let (sid, pid, oid) = (
543        const_subject(&t.0, dict)?,
544        const_predicate(&t.1)?,
545        const_object(&t.2, dict)?,
546    );
547    // Stream the matches with the lazy cursor — the hash join is
548    // order-independent, so no canonical re-sort is needed here.
549    for triple in index.scan_iter((sid, pid, oid)) {
550        if let Some(row) = triple_row(ctx, t, triple) {
551            rel.push(row);
552        }
553    }
554    Some((rel, slots))
555}
556
557/// A per-pattern cardinality estimate, smaller = cheaper to scan. The base is
558/// the **exact** per-predicate triple count (sum of the summary quotient graph's
559/// super-edge counts) when the predicate is bound, else the whole-graph total.
560/// A bound subject/object (a constant, or a variable already bound by `seed`)
561/// then scales it down by that predicate's **measured selectivity** from the
562/// `query_stats` block — `1 / distinct_subjects` for a bound subject (so
563/// `<s> <p> ?o` ≈ the average objects per subject; exactly 1 for a functional
564/// predicate), `1 / distinct_objects` for a bound object. Files built before the
565/// `query_stats` block fall back to fixed default selectivities. `None` when the
566/// file carries no pyramid summary at all — ordering then uses the constant-count
567/// heuristic.
568fn pattern_estimates(
569    ctx: &Ctx,
570    lowered: &[(SlotTerm, SlotTerm, SlotTerm)],
571    seed: &std::collections::HashSet<usize>,
572) -> Option<Vec<f64>> {
573    // Only use the summary when it's already in memory — never fault it just to
574    // plan (the lazy remote path defers the pyramid by design).
575    let pyr = ctx.rete.pyramid_if_loaded()?;
576    let mut pred: std::collections::HashMap<u32, u64> = std::collections::HashMap::new();
577    for e in &pyr.summary {
578        *pred.entry(e.predicate).or_insert(0) += e.count as u64;
579    }
580    // Measured per-predicate distinct subjects/objects (query_stats block; empty
581    // on files built before it existed → fall back to the default selectivities).
582    let stats: std::collections::HashMap<u32, &crate::meta::PredStat> = pyr
583        .predicate_stats
584        .iter()
585        .map(|s| (s.predicate, s))
586        .collect();
587    let total = ctx.rete.header().quad_count.max(1) as f64;
588    let num_preds = pred.len().max(1) as f64;
589    // Defaults used only when query_stats has no entry for the predicate.
590    const SEL_SUBJECT: f64 = 0.001;
591    const SEL_OBJECT: f64 = 0.02;
592    let node_bound = |t: &SlotTerm| match t {
593        SlotTerm::Node(_) => true,
594        SlotTerm::Var(v) => seed.contains(v),
595        SlotTerm::Pred(_) => false,
596    };
597    Some(
598        lowered
599            .iter()
600            .map(|t| {
601                // Base from the predicate: exact total for a constant predicate,
602                // the average predicate total for a seed-bound predicate variable
603                // (the value is known to the probe but not at plan time), else the
604                // whole graph. `st` is the measured stats for a constant predicate.
605                let (base, st) = match t.1 {
606                    SlotTerm::Pred(p) => {
607                        (*pred.get(&p).unwrap_or(&0) as f64, stats.get(&p).copied())
608                    }
609                    SlotTerm::Var(v) if seed.contains(&v) => (total / num_preds, None),
610                    _ => (total, None),
611                };
612                let mut est = base.max(1.0);
613                if node_bound(&t.0) {
614                    est *= match st {
615                        Some(s) if s.distinct_subjects > 0 => 1.0 / s.distinct_subjects as f64,
616                        _ => SEL_SUBJECT,
617                    };
618                }
619                if node_bound(&t.2) {
620                    est *= match st {
621                        Some(s) if s.distinct_objects > 0 => 1.0 / s.distinct_objects as f64,
622                        _ => SEL_OBJECT,
623                    };
624                }
625                est.max(1.0)
626            })
627            .collect(),
628    )
629}
630
631/// The minimum, over a BGP's patterns, of [`pattern_scan_bytes`] — the
632/// cheapest single-pattern scan the hash-join path would at least perform to
633/// materialize this BGP. `None` = a constant term is unknown to the
634/// dictionary (the BGP is unsatisfiable; the caller short-circuits anyway).
635/// Unlike [`pattern_estimates`], this needs no pyramid summary, so it works on
636/// remote-lazy files — exactly where the answer matters most.
637pub(crate) fn bgp_min_scan_bytes(
638    ctx: &Ctx,
639    index: &GraphIndex,
640    patterns: &[TriplePattern],
641) -> Option<u64> {
642    let lowered = lower(patterns, ctx)?;
643    lowered.iter().map(|t| pattern_scan_bytes(index, t)).min()
644}
645
646/// One pattern's cheapest scan extent in BYTES: for each permutation, sum the
647/// encoded lengths of the tiles a scan of this pattern would route to (all
648/// tiles covering the bound leading key, or the whole section when the leading
649/// key is unbound), and take the minimum. Tile *counts* are useless here — a
650/// broad predicate can sit inside one giant tile — but the directory's byte
651/// lengths expose it. Reads only resident metadata, never tile data.
652fn pattern_scan_bytes(index: &GraphIndex, t: &(SlotTerm, SlotTerm, SlotTerm)) -> u64 {
653    let sections = index.tile_sections();
654    let comp = |role: usize| match role {
655        0 => &t.0,
656        1 => &t.1,
657        _ => &t.2,
658    };
659    let mut best = u64::MAX;
660    for perm in crate::index::ALL_PERMS {
661        let tiles = sections[perm.section_index()];
662        best = best.min(match comp(perm.roles()[0]) {
663            SlotTerm::Node(id) | SlotTerm::Pred(id) => tiles
664                .iter()
665                .filter(|tile| {
666                    let (lo, hi) = tile.leading_range();
667                    lo <= *id && *id <= hi
668                })
669                .map(Tile::encoded_len)
670                .sum(),
671            SlotTerm::Var(_) => tiles.iter().map(Tile::encoded_len).sum(),
672        });
673    }
674    best
675}
676
677/// Scan extent above which a join side counts as "fat": materializing (or, on
678/// a remote index, fetching) ≥2 MiB of encoded tiles to build a hash/merge
679/// side is past the point where probing it per surviving row wins. Shared by
680/// the merge-seed asymmetry gate here and the left-join force-probe gate in
681/// `sparql::eval`.
682pub(crate) const FAT_SCAN_BYTES: u64 = 2 << 20;
683
684/// Order pattern indices for a left-deep join: cheapest (smallest estimated
685/// cardinality) first, always preferring a pattern that shares a variable with
686/// the already-joined set so the join stays connected and intermediate relations
687/// stay small. Cardinality comes from [`pattern_estimates`]; with no summary it
688/// falls back to a most-constants-first heuristic. Pure reordering — `hash_join`
689/// is order-independent, so the result multiset is unchanged.
690fn selectivity_order(ctx: &Ctx, lowered: &[(SlotTerm, SlotTerm, SlotTerm)]) -> Vec<usize> {
691    selectivity_order_seeded(ctx, lowered, &std::collections::HashSet::new())
692}
693
694/// [`selectivity_order`] with a set of slots already bound by an outer seed
695/// row: those variables count as already connected (and, in the no-summary
696/// fallback, as constants for selectivity).
697fn selectivity_order_seeded(
698    ctx: &Ctx,
699    lowered: &[(SlotTerm, SlotTerm, SlotTerm)],
700    seed: &std::collections::HashSet<usize>,
701) -> Vec<usize> {
702    let estimates = pattern_estimates(ctx, lowered, seed);
703    let consts = |t: &(SlotTerm, SlotTerm, SlotTerm)| {
704        [&t.0, &t.1, &t.2]
705            .into_iter()
706            .filter(|x| match x {
707                SlotTerm::Var(v) => seed.contains(v),
708                _ => true,
709            })
710            .count()
711    };
712    let vars = |t: &(SlotTerm, SlotTerm, SlotTerm)| -> Vec<usize> {
713        [&t.0, &t.1, &t.2]
714            .into_iter()
715            .filter_map(|x| match x {
716                SlotTerm::Var(v) => Some(*v),
717                _ => None,
718            })
719            .collect()
720    };
721    // Higher score = picked sooner. With summary stats that's the negated
722    // estimate (smaller cardinality wins); without, the constant count (old
723    // behaviour, byte-identical ordering for pyramid-less files).
724    let score = |i: usize| -> f64 {
725        match &estimates {
726            Some(e) => -e[i],
727            None => consts(&lowered[i]) as f64,
728        }
729    };
730    // A `?s rdf:type <Class>` with an as-yet-unbound subject is a **class
731    // enumeration**: it matches every instance of the class, so it is almost never
732    // the most selective place to *start* a join (a popular class has thousands of
733    // instances), even though it has two bound positions and so looks selective to
734    // the constant-count heuristic. Deprioritize it as a seed; once its subject is
735    // bound by another pattern it becomes a cheap per-row type *check* (the hybrid
736    // join probes it), so pushing it later costs nothing. Robust and stats-free.
737    let type_pid = ctx.rete.dictionary().predicate_id(crate::file::RDF_TYPE);
738    let is_class_enum = |i: usize, bound: &std::collections::HashSet<usize>| -> bool {
739        let t = &lowered[i];
740        type_pid.is_some_and(|tp| matches!(t.1, SlotTerm::Pred(p) if p == tp))
741            && matches!(t.2, SlotTerm::Node(_))
742            && matches!(t.0, SlotTerm::Var(v) if !bound.contains(&v))
743    };
744    let n = lowered.len();
745    let mut remaining: Vec<usize> = (0..n).collect();
746    let mut order: Vec<usize> = Vec::with_capacity(n);
747    let mut bound: std::collections::HashSet<usize> = seed.clone();
748    while !remaining.is_empty() {
749        // Pick the remaining pattern with the best (connected, not-a-class-enum,
750        // score) key; ties go to the lowest original index for a stable order.
751        let best = *remaining
752            .iter()
753            .max_by(|&&a, &&b| {
754                let connected = |i: usize| vars(&lowered[i]).iter().any(|v| bound.contains(v));
755                connected(a)
756                    .cmp(&connected(b))
757                    .then_with(|| is_class_enum(b, &bound).cmp(&is_class_enum(a, &bound)))
758                    .then_with(|| {
759                        score(a)
760                            .partial_cmp(&score(b))
761                            .unwrap_or(std::cmp::Ordering::Equal)
762                    })
763                    .then_with(|| b.cmp(&a))
764            })
765            .unwrap();
766        for v in vars(&lowered[best]) {
767            bound.insert(v);
768        }
769        order.push(best);
770        remaining.retain(|&i| i != best);
771    }
772    order
773}
774
775/// Does the BGP have at least one solution against `index`? A single pattern
776/// with all-distinct variables streams the index and stops at the first match
777/// (no materialization); anything else falls back to the full evaluator and
778/// tests non-emptiness (still benefiting from the lazy per-pattern scan). Used
779/// by `ASK`.
780pub(crate) fn bgp_exists(ctx: &Ctx, index: &GraphIndex, patterns: &[TriplePattern]) -> bool {
781    let dict = ctx.rete.dictionary();
782    let Some(lowered) = lower(patterns, ctx) else {
783        return false;
784    };
785    if let [t] = lowered.as_slice() {
786        // The fast path can't enforce a variable repeated across positions
787        // (e.g. `?x p ?x`) — that needs the row builder — so only take it when
788        // the pattern's variables are all distinct.
789        let names: Vec<usize> = [&t.0, &t.1, &t.2]
790            .into_iter()
791            .filter_map(|x| match x {
792                SlotTerm::Var(v) => Some(*v),
793                _ => None,
794            })
795            .collect();
796        let distinct = names
797            .iter()
798            .enumerate()
799            .all(|(i, v)| !names[i + 1..].contains(v));
800        if distinct {
801            return match (
802                const_subject(&t.0, dict),
803                const_predicate(&t.1),
804                const_object(&t.2, dict),
805            ) {
806                (Some(s), Some(p), Some(o)) => index.scan_iter((s, p, o)).next().is_some(),
807                _ => false,
808            };
809        }
810    }
811    !eval_bgp_rows(ctx, index, patterns).is_empty()
812}
813
814/// A lazy left-deep BGP join that yields solution rows one at a time, so a
815/// consumer under `LIMIT`/`OFFSET` (or `.next()` for ASK) can stop early —
816/// stopping also stops the underlying index scan.
817///
818/// The all-but-last patterns are joined eagerly into a `prefix` **hash table**
819/// (they are the most selective, so this is the small side); the *last* (least
820/// selective) pattern is then **streamed** from the index cursor, probing the
821/// prefix — the big scan is never materialized, and a `LIMIT` above stops it
822/// after a handful of triples. Yields the same solution multiset as
823/// [`eval_bgp_rows`] (the join is order-independent), just incrementally.
824pub(crate) struct BgpSolutions<'q> {
825    scan: Option<Box<dyn Iterator<Item = Row> + 'q>>,
826    prefix: Vec<Row>,
827    /// Prefix row indices keyed by the shared-slot values.
828    buckets: HashMap<Vec<Val>, Vec<usize>>,
829    shared: Vec<usize>,
830    cartesian: bool,
831    /// The prefix is the all-unbound seed row (single-pattern BGP): scanned
832    /// rows are already complete solutions — pass them through unmerged.
833    seed_only: bool,
834    cur_scan: Option<Row>,
835    /// Reused candidate buffer (avoids a Vec allocation per scanned row).
836    matches: Vec<usize>,
837    mi: usize,
838}
839
840impl<'q> BgpSolutions<'q> {
841    /// An iterator that yields nothing (an unsatisfiable BGP).
842    fn empty() -> Self {
843        BgpSolutions {
844            scan: None,
845            prefix: Vec::new(),
846            buckets: HashMap::new(),
847            shared: Vec::new(),
848            cartesian: false,
849            seed_only: false,
850            cur_scan: None,
851            matches: Vec::new(),
852            mi: 0,
853        }
854    }
855
856    pub(crate) fn new(ctx: &'q Ctx<'q>, index: &'q GraphIndex, patterns: &[TriplePattern]) -> Self {
857        // An empty BGP has exactly one (empty) solution.
858        if patterns.is_empty() {
859            return BgpSolutions {
860                scan: Some(Box::new(std::iter::once(ctx.slots.empty_row()))),
861                prefix: vec![ctx.slots.empty_row()],
862                buckets: HashMap::new(),
863                shared: Vec::new(),
864                cartesian: true,
865                seed_only: true,
866                cur_scan: None,
867                matches: Vec::new(),
868                mi: 0,
869            };
870        }
871        // Lower all patterns; an unknown constant term ⇒ no solutions.
872        let Some(lowered) = lower(patterns, ctx) else {
873            return Self::empty();
874        };
875        // Join all but the *last* (least selective) pattern eagerly into the
876        // prefix; a single pattern leaves the seed row as the prefix.
877        let order = selectivity_order(ctx, &lowered);
878        let (&last_i, prefix_is) = order.split_last().unwrap();
879        let prefix_pats: Vec<TriplePattern> =
880            prefix_is.iter().map(|&i| patterns[i].clone()).collect();
881        let prefix = eval_bgp_rows(ctx, index, &prefix_pats);
882        if prefix.is_empty() {
883            return Self::empty();
884        }
885
886        // Stream the last pattern with the lazy index cursor.
887        let Some(scan) = scan_rows(ctx, index, lowered[last_i]) else {
888            return Self::empty();
889        };
890
891        // Hash-join key = slots shared by the prefix and the last pattern (BGP
892        // rows bind every slot of their pattern set, so the prefix's bound set
893        // can be read off its first row).
894        let shared: Vec<usize> = pattern_slots(&lowered[last_i])
895            .into_iter()
896            .filter(|&s| prefix[0][s].is_some())
897            .collect();
898        let cartesian = shared.is_empty();
899        // A single-pattern BGP joins against the all-unbound seed: scanned rows
900        // are already complete solutions.
901        let seed_only = prefix.len() == 1 && prefix[0].iter().all(Option::is_none);
902        let mut buckets: HashMap<Vec<Val>, Vec<usize>> = HashMap::new();
903        if !cartesian {
904            for (i, r) in prefix.iter().enumerate() {
905                let key: Vec<Val> = shared.iter().map(|&s| r[s].clone().unwrap()).collect();
906                buckets.entry(key).or_default().push(i);
907            }
908        }
909        BgpSolutions {
910            scan: Some(Box::new(scan)),
911            prefix,
912            buckets,
913            shared,
914            cartesian,
915            seed_only,
916            cur_scan: None,
917            matches: Vec::new(),
918            mi: 0,
919        }
920    }
921}
922
923impl Iterator for BgpSolutions<'_> {
924    type Item = Row;
925
926    fn next(&mut self) -> Option<Row> {
927        // Single-pattern fast path: pass scanned rows through unmerged.
928        if self.seed_only {
929            return self.scan.as_mut()?.next();
930        }
931        loop {
932            // Emit the next prefix match for the current scanned row.
933            if self.mi < self.matches.len() {
934                let pi = self.matches[self.mi];
935                self.mi += 1;
936                let mut merged = self.prefix[pi].clone();
937                for (slot, v) in self.cur_scan.as_ref().unwrap().iter().enumerate() {
938                    if v.is_some() {
939                        merged[slot] = v.clone();
940                    }
941                }
942                return Some(merged);
943            }
944            // Pull the next row from the index scan and gather its matches
945            // (into the reused buffer — no allocation per scanned row).
946            let s = self.scan.as_mut()?.next()?;
947            self.matches.clear();
948            if self.cartesian {
949                self.matches.extend(0..self.prefix.len());
950            } else {
951                let key: Vec<Val> = self.shared.iter().map(|&i| s[i].clone().unwrap()).collect();
952                if let Some(c) = self.buckets.get(&key) {
953                    self.matches.extend_from_slice(c);
954                }
955            }
956            self.mi = 0;
957            self.cur_scan = Some(s);
958        }
959    }
960}
961
962// --- index-nested-loop probing ------------------------------------------------
963//
964// Under a small, known demand (LIMIT/ASK — see `Ctx::limit_hint`), scanning
965// every pattern once to hash-join is mostly wasted work: the consumer wants a
966// handful of rows. The probe path instead streams the seed/first pattern and,
967// per row, *probes* each next pattern through the index with the row's bound
968// values substituted as scan constants — so producing k solutions touches
969// O(k · patterns) index groups instead of every pattern's full extent. Same
970// solution multiset as the hash join (joins are order-independent); only the
971// evaluation order differs.
972
973/// Index-scan constant for a subject position given a partially-bound row.
974/// Outer `None` = unsatisfiable for this row; inner `None` = wildcard.
975fn probe_subject(ctx: &Ctx, t: &SlotTerm, base: &Row) -> Option<Option<u32>> {
976    let dict = ctx.rete.dictionary();
977    match t {
978        SlotTerm::Node(n) => dict.node_as_subject_id(*n).map(Some),
979        SlotTerm::Pred(_) => None,
980        SlotTerm::Var(i) => match &base[*i] {
981            None => Some(None),
982            Some(Val::Id(v)) if *v >= 0 => dict.node_as_subject_id(*v as u32).map(Some),
983            // A predicate-tagged or computed value can never be a subject.
984            Some(_) => None,
985        },
986    }
987}
988
989/// Index-scan constant for an object position (see [`probe_subject`]).
990fn probe_object(ctx: &Ctx, t: &SlotTerm, base: &Row) -> Option<Option<u32>> {
991    let dict = ctx.rete.dictionary();
992    match t {
993        SlotTerm::Node(n) => dict.node_as_object_id(*n).map(Some),
994        SlotTerm::Pred(_) => None,
995        SlotTerm::Var(i) => match &base[*i] {
996            None => Some(None),
997            Some(Val::Id(v)) if *v >= 0 => dict.node_as_object_id(*v as u32).map(Some),
998            Some(_) => None,
999        },
1000    }
1001}
1002
1003/// Index-scan constant for a predicate position (see [`probe_subject`]).
1004fn probe_predicate(ctx: &Ctx, t: &SlotTerm, base: &Row) -> Option<Option<u32>> {
1005    let dict = ctx.rete.dictionary();
1006    match t {
1007        SlotTerm::Pred(p) => Some(Some(*p)),
1008        SlotTerm::Node(_) => None,
1009        SlotTerm::Var(i) => match &base[*i] {
1010            None => Some(None),
1011            Some(Val::Id(v)) if *v < 0 => Some(Some((-v - 1) as u32)),
1012            // Canonicalized to a node id — the term may still be a predicate.
1013            Some(Val::Id(v)) => ctx
1014                .resolver
1015                .term(*v)
1016                .and_then(|t| dict.predicate_id(&t))
1017                .map(Some),
1018            Some(Val::Str(_)) => None,
1019        },
1020    }
1021}
1022
1023/// Probe one pattern with a partially-bound row: every bound variable becomes
1024/// an index-scan constant, and each matching triple extends a clone of the row
1025/// (repeated unbound variables stay consistent).
1026fn probe_rows<'q>(
1027    ctx: &'q Ctx<'q>,
1028    index: &'q GraphIndex,
1029    t: (SlotTerm, SlotTerm, SlotTerm),
1030    base: Row,
1031) -> Box<dyn Iterator<Item = Row> + 'q> {
1032    let (Some(sid), Some(pid), Some(oid)) = (
1033        probe_subject(ctx, &t.0, &base),
1034        probe_predicate(ctx, &t.1, &base),
1035        probe_object(ctx, &t.2, &base),
1036    ) else {
1037        return Box::new(std::iter::empty());
1038    };
1039    let dict = ctx.rete.dictionary();
1040    Box::new(
1041        index
1042            .scan_iter((sid, pid, oid))
1043            .filter_map(move |(s_id, p_id, o_id)| {
1044                let s_val = dict.subject_node(s_id) as i64;
1045                let p_val = ctx.resolver.canon_id(pred_tag(p_id));
1046                let o_val = dict.object_node(o_id) as i64;
1047                let mut row = base.clone();
1048                for (term, val) in [(&t.0, s_val), (&t.1, p_val), (&t.2, o_val)] {
1049                    if let SlotTerm::Var(i) = term {
1050                        match row[*i] {
1051                            Some(Val::Id(existing)) if existing != val => return None,
1052                            Some(Val::Id(_)) => {}
1053                            Some(Val::Str(_)) => return None,
1054                            None => row[*i] = Some(Val::Id(val)),
1055                        }
1056                    }
1057                }
1058                Some(row)
1059            }),
1060    )
1061}
1062
1063/// A lowered, probe-ordered BGP, reusable across many seed rows.
1064pub(crate) struct ProbePlan {
1065    pats: Vec<(SlotTerm, SlotTerm, SlotTerm)>,
1066}
1067
1068impl ProbePlan {
1069    /// Lower and order `patterns` for probing from rows that bind (at least)
1070    /// the slots in `seed_mask`. `None` = a constant term is unknown, making
1071    /// the BGP unsatisfiable for every seed.
1072    pub(crate) fn new(ctx: &Ctx, patterns: &[TriplePattern], seed_mask: &[bool]) -> Option<Self> {
1073        let lowered = lower(patterns, ctx)?;
1074        let seed: std::collections::HashSet<usize> = seed_mask
1075            .iter()
1076            .enumerate()
1077            .filter_map(|(i, b)| b.then_some(i))
1078            .collect();
1079        let order = selectivity_order_seeded(ctx, &lowered, &seed);
1080        Some(ProbePlan {
1081            pats: order.into_iter().map(|i| lowered[i]).collect(),
1082        })
1083    }
1084}
1085
1086/// Depth-first index-nested-loop join over a [`ProbePlan`], starting from a
1087/// seed row. Fully lazy: pulling k rows probes O(k · patterns) index groups.
1088pub(crate) struct ProbeJoin<'q> {
1089    ctx: &'q Ctx<'q>,
1090    index: &'q GraphIndex,
1091    pats: Vec<(SlotTerm, SlotTerm, SlotTerm)>,
1092    stack: Vec<Box<dyn Iterator<Item = Row> + 'q>>,
1093}
1094
1095impl<'q> ProbeJoin<'q> {
1096    /// Probe a whole BGP from scratch (the seed is the all-unbound row).
1097    /// `None` = unsatisfiable.
1098    pub(crate) fn new(
1099        ctx: &'q Ctx<'q>,
1100        index: &'q GraphIndex,
1101        patterns: &[TriplePattern],
1102    ) -> Option<Self> {
1103        let plan = ProbePlan::new(ctx, patterns, &vec![false; ctx.slots.len()])?;
1104        Some(Self::from_plan(ctx, index, &plan, ctx.slots.empty_row()))
1105    }
1106
1107    /// Probe a pre-lowered plan from one seed row.
1108    pub(crate) fn from_plan(
1109        ctx: &'q Ctx<'q>,
1110        index: &'q GraphIndex,
1111        plan: &ProbePlan,
1112        seed: Row,
1113    ) -> Self {
1114        let pats = plan.pats.clone();
1115        let first = probe_rows(ctx, index, pats[0], seed);
1116        ProbeJoin {
1117            ctx,
1118            index,
1119            pats,
1120            stack: vec![first],
1121        }
1122    }
1123}
1124
1125impl Iterator for ProbeJoin<'_> {
1126    type Item = Row;
1127
1128    fn next(&mut self) -> Option<Row> {
1129        loop {
1130            let depth = self.stack.len();
1131            match self.stack.last_mut()?.next() {
1132                Some(row) => {
1133                    if depth == self.pats.len() {
1134                        return Some(row);
1135                    }
1136                    let it = probe_rows(self.ctx, self.index, self.pats[depth], row);
1137                    self.stack.push(it);
1138                }
1139                None => {
1140                    self.stack.pop();
1141                    if self.stack.is_empty() {
1142                        return None;
1143                    }
1144                }
1145            }
1146        }
1147    }
1148}
1149
1150/// Hash-join two BGP relations on the slots they share. Every left row binds
1151/// exactly the slots accumulated so far and every right row binds the current
1152/// pattern's slots, so the shared set is uniform across rows.
1153fn hash_join(
1154    left: Vec<Row>,
1155    left_bound: &[usize],
1156    right: Vec<Row>,
1157    right_slots: &[usize],
1158) -> Vec<Row> {
1159    // Fast path: the all-unbound seed row joins to `right` unchanged.
1160    if left.len() == 1 && left_bound.is_empty() {
1161        return right;
1162    }
1163    if left.is_empty() || right.is_empty() {
1164        return Vec::new();
1165    }
1166    let shared: Vec<usize> = right_slots
1167        .iter()
1168        .copied()
1169        .filter(|s| left_bound.contains(s))
1170        .collect();
1171    let fill = |l: &Row, r: &Row| -> Row {
1172        let mut out = l.clone();
1173        for &s in right_slots {
1174            out[s] = r[s].clone();
1175        }
1176        out
1177    };
1178    if shared.is_empty() {
1179        // No shared variable: Cartesian product. In a connected BGP this only
1180        // arises for a fully-ground pattern (deduped index ⇒ one matching row),
1181        // so it acts as an existence filter rather than multiplying rows.
1182        let mut out = Vec::with_capacity(left.len() * right.len());
1183        for l in &left {
1184            for r in &right {
1185                out.push(fill(l, r));
1186            }
1187        }
1188        return out;
1189    }
1190    let key_of = |b: &Row| -> Vec<Val> { shared.iter().map(|&s| b[s].clone().unwrap()).collect() };
1191    // A hash join is symmetric on the shared key, so build the **smaller** side's
1192    // hash table and probe with the larger — same result set, fewer inserts and a
1193    // smaller table. (The old code always built `right`; after cardinality-based
1194    // ordering the accumulating `left` is usually the small selective side, so
1195    // building it instead is the common win.) Whichever side is built, the output
1196    // row carries every bound slot of both.
1197    let mut out = Vec::new();
1198    if right.len() <= left.len() {
1199        let mut buckets: HashMap<Vec<Val>, Vec<Row>> = HashMap::new();
1200        for r in right {
1201            buckets.entry(key_of(&r)).or_default().push(r);
1202        }
1203        for l in &left {
1204            if let Some(rs) = buckets.get(&key_of(l)) {
1205                for r in rs {
1206                    out.push(fill(l, r));
1207                }
1208            }
1209        }
1210    } else {
1211        // Build the left side; fill the left's own slots into each probed right.
1212        let mut buckets: HashMap<Vec<Val>, Vec<Row>> = HashMap::new();
1213        for l in left {
1214            buckets.entry(key_of(&l)).or_default().push(l);
1215        }
1216        for r in &right {
1217            if let Some(ls) = buckets.get(&key_of(r)) {
1218                for l in ls {
1219                    let mut row = r.clone();
1220                    for &s in left_bound {
1221                        if let Some(v) = &l[s] {
1222                            row[s] = Some(v.clone());
1223                        }
1224                    }
1225                    out.push(row);
1226                }
1227            }
1228        }
1229    }
1230    out
1231}
1232
1233// Constant-only index constraints. A variable scans as a wildcard (`Some(None)`)
1234// — it is resolved later by the hash join, not pushed into the scan. The outer
1235// `None` means "unsatisfiable" (a constant in an impossible role, or unknown to
1236// the dictionary), which empties the whole BGP. Inner `None` = "wildcard".
1237fn const_subject(t: &SlotTerm, d: &crate::Dictionary) -> Option<Option<u32>> {
1238    match t {
1239        SlotTerm::Node(n) => d.node_as_subject_id(*n).map(Some),
1240        SlotTerm::Pred(_) => None, // a predicate term can't be a subject
1241        SlotTerm::Var(_) => Some(None),
1242    }
1243}
1244
1245fn const_object(t: &SlotTerm, d: &crate::Dictionary) -> Option<Option<u32>> {
1246    match t {
1247        SlotTerm::Node(n) => d.node_as_object_id(*n).map(Some),
1248        SlotTerm::Pred(_) => None,
1249        SlotTerm::Var(_) => Some(None),
1250    }
1251}
1252
1253fn const_predicate(t: &SlotTerm) -> Option<Option<u32>> {
1254    match t {
1255        SlotTerm::Pred(p) => Some(Some(*p)),
1256        SlotTerm::Node(_) => None, // a node term can't be a predicate
1257        SlotTerm::Var(_) => Some(None),
1258    }
1259}
1260
1261#[cfg(test)]
1262mod tests {
1263    use super::*;
1264    use crate::dictionary::DictionaryBuilder;
1265    use crate::file::write_file;
1266    use crate::index::GraphIndexBuilder;
1267
1268    fn rete_from(triples: &[(&str, &str, &str)]) -> Vec<u8> {
1269        let mut db = DictionaryBuilder::new();
1270        for (s, p, o) in triples {
1271            db.observe(s, p, o);
1272        }
1273        let dict = db.build();
1274        let mut ib = GraphIndexBuilder::new();
1275        for (s, p, o) in triples {
1276            ib.push(dict.encode(s, p, o).unwrap());
1277        }
1278        write_file(&dict, &ib.build(), false, &[], 0)
1279    }
1280
1281    fn pat(s: &str, p: &str, o: &str) -> TriplePattern {
1282        TriplePattern {
1283            s: PatternTerm::parse(s),
1284            p: PatternTerm::parse(p),
1285            o: PatternTerm::parse(o),
1286        }
1287    }
1288
1289    #[test]
1290    fn single_pattern_binds_variable() {
1291        let bytes = rete_from(&[("Alice", "knows", "Bob"), ("Bob", "knows", "Carol")]);
1292        let rete = Rete::open(&bytes).unwrap();
1293        let sols = eval_bgp(&rete, &[pat("Alice", "knows", "?y")]);
1294        assert_eq!(sols.len(), 1);
1295        assert_eq!(sols[0]["y"], "Bob");
1296    }
1297
1298    #[test]
1299    fn two_hop_join_on_shared_variable() {
1300        // Alice -> Bob -> Carol, plus a dead-end branch.
1301        let bytes = rete_from(&[
1302            ("Alice", "knows", "Bob"),
1303            ("Bob", "knows", "Carol"),
1304            ("Carol", "knows", "Dave"),
1305            ("Alice", "knows", "Eve"), // Eve knows no one -> no 2-hop
1306        ]);
1307        let rete = Rete::open(&bytes).unwrap();
1308        let sols = eval_bgp(&rete, &[pat("?x", "knows", "?y"), pat("?y", "knows", "?z")]);
1309        // Alice-Bob-Carol, Bob-Carol-Dave.
1310        let mut got: Vec<_> = sols
1311            .iter()
1312            .map(|b| (b["x"].clone(), b["y"].clone(), b["z"].clone()))
1313            .collect();
1314        got.sort();
1315        assert_eq!(
1316            got,
1317            vec![
1318                ("Alice".into(), "Bob".into(), "Carol".into()),
1319                ("Bob".into(), "Carol".into(), "Dave".into()),
1320            ]
1321        );
1322    }
1323
1324    #[test]
1325    fn repeated_variable_within_pattern() {
1326        // A mutual/self relation: only Bob knows himself.
1327        let bytes = rete_from(&[("Alice", "knows", "Bob"), ("Bob", "knows", "Bob")]);
1328        let rete = Rete::open(&bytes).unwrap();
1329        let sols = eval_bgp(&rete, &[pat("?x", "knows", "?x")]);
1330        assert_eq!(sols.len(), 1);
1331        assert_eq!(sols[0]["x"], "Bob");
1332    }
1333
1334    #[test]
1335    fn no_solutions_yields_empty() {
1336        let bytes = rete_from(&[("Alice", "knows", "Bob")]);
1337        let rete = Rete::open(&bytes).unwrap();
1338        assert!(eval_bgp(&rete, &[pat("Alice", "likes", "?y")]).is_empty());
1339    }
1340
1341    /// A LUBM-Q7-shaped snowflake: a popular class enumeration (`?x a Student`)
1342    /// plus a selective adjacency seed (`<Prof> teacherOf ?y`). Exercises both the
1343    /// **class-enum seed deprioritization** (the join must not start from the huge
1344    /// `a Student` extent) and the **probe-the-tail hybrid** (the type checks are
1345    /// applied late, per row). The answer must equal a brute-force computation
1346    /// regardless of how the join is ordered/probed.
1347    #[test]
1348    fn snowflake_type_and_adjacency_matches_reference() {
1349        const TYPE: &str = "<http://www.w3.org/1999/02/22-rdf-syntax-ns#type>";
1350        let mut triples: Vec<(String, String, String)> = Vec::new();
1351        // 60 students, each a Student and taking 2 courses (c{i%5}, c{(i+1)%5}).
1352        for i in 0..60 {
1353            let s = format!("S{i}");
1354            triples.push((s.clone(), TYPE.into(), "Student".into()));
1355            triples.push((s.clone(), "takesCourse".into(), format!("c{}", i % 5)));
1356            triples.push((s, "takesCourse".into(), format!("c{}", (i + 1) % 5)));
1357        }
1358        // 5 courses; the professor teaches only c0 and c2.
1359        for j in 0..5 {
1360            triples.push((format!("c{j}"), TYPE.into(), "Course".into()));
1361        }
1362        triples.push(("Prof".into(), "teacherOf".into(), "c0".into()));
1363        triples.push(("Prof".into(), "teacherOf".into(), "c2".into()));
1364
1365        let refs: Vec<(&str, &str, &str)> = triples
1366            .iter()
1367            .map(|(s, p, o)| (s.as_str(), p.as_str(), o.as_str()))
1368            .collect();
1369        let bytes = rete_from(&refs);
1370        let rete = Rete::open(&bytes).unwrap();
1371
1372        let got: std::collections::BTreeSet<(String, String)> = eval_bgp(
1373            &rete,
1374            &[
1375                pat("?x", TYPE, "Student"),
1376                pat("?y", TYPE, "Course"),
1377                pat("?x", "takesCourse", "?y"),
1378                pat("Prof", "teacherOf", "?y"),
1379            ],
1380        )
1381        .into_iter()
1382        .map(|b| (b["x"].clone(), b["y"].clone()))
1383        .collect();
1384
1385        // Brute force: students taking a course the professor teaches (c0 or c2).
1386        let taught = ["c0", "c2"];
1387        let mut want = std::collections::BTreeSet::new();
1388        for i in 0..60 {
1389            for c in [i % 5, (i + 1) % 5] {
1390                let course = format!("c{c}");
1391                if taught.contains(&course.as_str()) {
1392                    want.insert((format!("S{i}"), course));
1393                }
1394            }
1395        }
1396        assert_eq!(got, want, "snowflake join must match the brute-force set");
1397        assert!(!want.is_empty(), "sanity: the reference set is non-empty");
1398    }
1399
1400    /// A subject-subject join (`?x a ?av . ?x b ?bv`) is the same-role shape the
1401    /// merge join handles: both patterns sort on the subject column. With multiple
1402    /// `a` and `b` values per subject it must emit the full cross product per
1403    /// subject (equal-key run handling) and drop subjects missing either side.
1404    #[test]
1405    fn merge_join_subject_star_cross_product() {
1406        let bytes = rete_from(&[
1407            ("x", "a", "a1"),
1408            ("x", "a", "a2"),
1409            ("x", "b", "b1"),
1410            ("x", "b", "b2"),
1411            ("y", "a", "a3"),
1412            ("y", "b", "b3"),
1413            ("z", "a", "a4"), // z has no `b` → contributes no join row
1414        ]);
1415        let rete = Rete::open(&bytes).unwrap();
1416        let mut got: Vec<(String, String, String)> =
1417            eval_bgp(&rete, &[pat("?x", "a", "?av"), pat("?x", "b", "?bv")])
1418                .iter()
1419                .map(|m| (m["x"].clone(), m["av"].clone(), m["bv"].clone()))
1420                .collect();
1421        got.sort();
1422        let mut want: Vec<(String, String, String)> = [
1423            ("x", "a1", "b1"),
1424            ("x", "a1", "b2"),
1425            ("x", "a2", "b1"),
1426            ("x", "a2", "b2"),
1427            ("y", "a3", "b3"),
1428        ]
1429        .into_iter()
1430        .map(|(a, b, c)| (a.to_string(), b.to_string(), c.to_string()))
1431        .collect();
1432        want.sort();
1433        assert_eq!(got, want, "same-role merge must equal the brute-force join");
1434    }
1435}