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};
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, index, &lowered);
149    // What a full scan of each pattern would produce (its own constants only),
150    // from the tile directory: the probe-vs-scan decision below compares it
151    // with the running result.
152    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    // Merge-join seed: if the two cheapest patterns share a same-role variable the
158    // 6-permutation index can co-sort them on, join them with a linear two-pointer
159    // **merge** (no hash table) and skip both below. Falls back silently to the
160    // hash/probe path when the shape doesn't qualify.
161    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; // already consumed by the merge seed
174        }
175        let t = lowered[idx];
176        // Hybrid join: once the running result is small and the next pattern shares
177        // an already-bound variable, **probe** it per row through the index instead
178        // of scanning its whole extent and hash-joining. A left-deep hash join
179        // otherwise materializes every pattern's full scan even when only a handful
180        // of rows survive — e.g. a `?x rdf:type C` over thousands of instances when
181        // the prefix already pinned ?x to 26 rows. Probing turns that O(scan) into
182        // O(rows × lookup). The full scan + hash join stays the path when the prefix
183        // is large (one scan beats many probes) or the pattern is a cartesian join
184        // (no shared bound variable to constrain the probe).
185        let shares_bound = pattern_slots(&t).iter().any(|s| bound.contains(s));
186        // Decide probe (per-row index lookup) vs scan + hash join. In memory a
187        // probe is a cheap lookup, so we probe up to a moderate prefix. Remotely
188        // each probe is a byte-range round-trip: probing a 500-row prefix is 500
189        // sequential fetches, while scanning a selectively-bound pattern is a few
190        // coalesced reads — so remotely we only probe a broad (whole-predicate)
191        // pattern, or a tiny prefix, and never a large one.
192        //
193        // Whatever the reader, probe when the scan would be far larger than
194        // the rows asking about it: 517 `?s a :Service` rows must not scan
195        // and materialize all 109 M `publicationYear` triples to keep 517.
196        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            // Over a lazy reader each probe is a byte-range round trip, and they run
209            // one at a time. First fault the tiles the *whole* batch will route to,
210            // together — one coalesced parallel read per section instead of N
211            // sequential ones — so the probes below hit a warm tile cache. No-op
212            // for a local index (every tile is already resident).
213            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
252/// Merge-join two base patterns that share **exactly one** variable sitting in the
253/// **same canonical position** (both subject, or both object) in each. That is the
254/// only shape rete's role-aware id spaces let a sort-merge align: the index sorts
255/// subject-ids and object-ids in independent spaces, so a cross-role (subject↔
256/// object) join's two streams are not co-sorted on the join value. Each side is
257/// streamed from the 6-permutation index already sorted on that column
258/// ([`GraphIndex::scan_iter_sorted_on`]), so the join is a linear two-pointer
259/// merge with no hash table — equal-key runs cross-product. Returns
260/// `(rows, bound_slots)`, or `None` when the shape/index can't support a merge and
261/// the caller should hash-join instead. The result multiset equals the hash
262/// join's (a join is order-independent).
263fn 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; // cross-role: the two role id-spaces aren't co-sorted
286    }
287    // A merge MATERIALIZES both sides in full (`collect` below builds a Vec per
288    // side). When one side is a pinpoint (a bound leading key routing to a few
289    // KB of tiles) and the other is fat (a broad predicate spanning megabytes),
290    // the linear merge costs building — and, on a remote index, FETCHING — the
291    // fat extent end-to-end: gigabytes over HTTP, and the 32-bit wasm OOM
292    // behind `?x wdt:P279 <C> . ?x rdfs:label ?l` on a 185M-triple remote
293    // graph. The hybrid loop handles that shape strictly better (scan the
294    // pinpoint side, probe the fat one per surviving row), so leave the merge
295    // seed to comparably-sized sides.
296    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    // Materialize each side as `(join-key, row)`, already ascending in the join
307    // column (the index streams it sorted there).
308    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                        // Combine the two rows; they agree on the shared slot `v`
339                        // (same key ⇒ same node value at the same column).
340                        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
370/// Probe the join tail (instead of full-scanning) once the running BGP result is
371/// at most this many rows: each remaining pattern that shares a bound variable is
372/// then resolved by `rows × index-lookup` rather than a whole-extent scan. Kept
373/// small so the probe count is always cheap in absolute terms — the dominant win
374/// is the large-scan-after-tiny-prefix case (a `?x a C` over thousands when the
375/// prefix already pinned ?x to a few dozen rows); a *moderately* large prefix
376/// keeps the one-pass scan + hash join, which beats thousands of probes.
377const BGP_PROBE_THRESHOLD: usize = 512;
378
379/// A pattern whose estimated scan is below this many triples is never worth
380/// planning around: scanning it is cheap whatever the other side holds. Keeps
381/// every small graph (the W3C suite, the playground datasets) on the joins it
382/// always used.
383const PROBE_SCAN_FLOOR: f64 = 65_536.0;
384
385/// Probe instead of scanning once the scan is at least this many times larger
386/// than the rows that would probe it. A probe is a tile-directory binary search
387/// plus a group lookup in an (often cached) tile; a scan pays per triple, and
388/// on the hash-join path also materializes every one of them.
389const PROBE_RATIO: f64 = 16.0;
390
391/// Is probing `rows` bound rows cheaper than scanning a pattern estimated at
392/// `scan_est` triples?
393fn probe_beats_scan(rows: usize, scan_est: f64) -> bool {
394    scan_est >= limits().probe_floor && (rows as f64) * PROBE_RATIO <= scan_est
395}
396
397/// The most rows that may probe a BGP whose cheapest pattern is estimated at
398/// `est` triples while probing still beats evaluating it whole
399/// ([`probe_beats_scan`]); `None` below the probe floor, where the BGP is
400/// cheap to evaluate either way.
401pub(crate) fn probe_row_budget(est: f64) -> Option<usize> {
402    (est >= limits().probe_floor).then(|| (est / PROBE_RATIO) as usize)
403}
404
405/// The smallest tile-directory estimate ([`GraphIndex::estimate_matches`])
406/// among a BGP's patterns — a floor on what evaluating it whole would scan.
407/// `None` when a constant is not in the file (the BGP matches nothing).
408pub(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
419/// The sum of the tile-directory estimates of a BGP's patterns: what evaluating
420/// it whole reads when each pattern is scanned once. [`bgp_min_estimate`] is
421/// only the cheapest pattern's share of that, so on a BGP of several mid-sized
422/// patterns (a journal's title, identifiers and their scheme: 38 k rows each
423/// under the probe floor, 5x that together) it stays under the floor and the
424/// BGP is materialized for a handful of left rows. `None` as for
425/// [`bgp_min_estimate`].
426pub(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
439/// Is `t` the pattern `?x rdf:type ?c` with `?x` among the slots `shared` with
440/// the rows probing it, and `?c` not?
441fn 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
450/// Is a BGP whose cheapest pattern is estimated at `est` rows fat enough that
451/// probing it per outer row beats loading it? [`FAT_SCAN_ROWS`]; tests set the
452/// probe floor instead, to drive the path over a graph small enough to check.
453fn 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
461/// A BGP probed once per outer row instead of evaluated whole: the shape of a
462/// correlated `FILTER (NOT) EXISTS` or `MINUS` over a BGP. Each outer row's
463/// bound variables become constants of the probe ([`ProbeJoin`]), so a fat
464/// BGP is never loaded; a row costs a few index lookups and stops at its
465/// first solution. A BGP below [`FAT_SCAN_ROWS`] is cheap to load once and
466/// index, so it keeps that plan: probing it would only fault scattered tiles.
467pub(crate) struct CorrelatedProbe {
468    /// `None`: a constant is not in the file, so no row has a solution.
469    plan: Option<ProbePlan>,
470    /// The variable slots the BGP mentions.
471    slots: Vec<usize>,
472}
473
474impl CorrelatedProbe {
475    /// `seed` marks the slots the outer rows certainly bind. `None` = evaluate
476    /// whole: the BGP is empty, shares no variable with the outer side, or is
477    /// not fat.
478    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        // An unknown constant empties the BGP for every row.
503        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    /// Does `row` bind a variable of the BGP?
516    pub(crate) fn shares(&self, row: &Row) -> bool {
517        self.slots.iter().any(|&i| row[i].is_some())
518    }
519
520    /// Does the BGP, with `row`'s bindings substituted, have a solution?
521    /// Callers check [`Self::shares`] first: a row that binds none of the
522    /// BGP's variables would re-evaluate all of it.
523    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/// Run `f` with the probe floor set to `floor` (tests only), so a join's
534/// probe paths can be driven over a graph small enough to check by hand.
535#[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/// The planner's size thresholds, gathered so tests can shrink them and drive
550/// every join path over a graph small enough to check by hand.
551#[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
575/// A BGP whose most selective pattern is estimated above this many matches is
576/// **streamed** rather than joined into a materialized prefix
577/// ([`BgpSolutions`]): every later pattern is then joined per row (probe) or
578/// against a small materialized pattern (hash), so resident memory no longer
579/// follows the largest pattern. `?s dcite:doi ?d . ?s dcite:fromSnapshot ?x`
580/// materialized 109 M rows before joining and ran a 16 GB container out of
581/// memory.
582const STREAM_SEED_ROWS: f64 = (1u64 << 21) as f64;
583
584/// On the streaming path, a later pattern estimated at no more than this many
585/// matches is scanned once into a hash table; a larger one is probed per row.
586const HASH_BUILD_ROWS: f64 = (1u64 << 20) as f64;
587
588/// Tile-directory cardinality estimates ([`GraphIndex::estimate_matches`]) for
589/// each lowered pattern. A variable an outer seed row binds is a constant at
590/// run time but unknown now: it scales the estimate by the same default
591/// selectivities [`pattern_estimates`] falls back to. An unsatisfiable
592/// constant estimates 0.
593fn 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    // A seeded subject/object of a constant predicate takes that predicate's
606    // MEASURED selectivity from `query_stats` (the SCHEMA section, kind 9, or
607    // a resident pyramid-meta) instead of the fixed default. The tile
608    // directory cannot see a value that only exists at run time, so this is
609    // the one estimate here the statistics improve. They are read only when
610    // some pattern needs them: one small range, once per session, on a lazy
611    // open — never for a query whose patterns have no seeded position.
612    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
662/// Remote (lazy) probe bounds. Over HTTP byte-range reads, a per-row probe is a
663/// network round-trip, so we probe far more reluctantly than in memory: only when
664/// scanning the next pattern would be a *broad* read (a whole-predicate scan,
665/// [`pattern_is_selective`] is false), or the prefix is tiny enough that a handful
666/// of probes is trivial either way. A pattern that carries its own bound
667/// subject/object routes to a narrow index range — cheaper to scan once and
668/// hash-join than to probe per prefix row. Probing a large prefix is always a
669/// loss remotely, so it is capped.
670const REMOTE_PROBE_MIN: usize = 8;
671const REMOTE_PROBE_MAX: usize = 1024;
672
673/// The remote probe budget, widened by the reader's concurrency: a serial
674/// sync-XHR reader (a phone without the COI pool) pays one full round trip per
675/// probe, so its budget stays at [`REMOTE_PROBE_MAX`]; a reader overlapping
676/// 16 range reads (the CLI's thread pool, the asyncified fetch variant, the
677/// COI fetch-worker pool) amortizes ~16 probes per round trip and gets a
678/// proportionally larger budget before the one-pass scan wins again.
679fn remote_probe_max(index: &GraphIndex) -> usize {
680    REMOTE_PROBE_MAX * index.read_concurrency().clamp(1, 16)
681}
682
683/// Batch-fault the tiles that a [`ProbePlan`]'s FIRST pattern will route to
684/// across a batch of seed rows — one coalesced read per section instead of a
685/// blocking round trip per row. Only the first pattern is prefetchable (later
686/// ones bind from earlier probe results), and only rows whose seed binds the
687/// pattern's subject or object are included: a predicate-only pattern routes
688/// to the predicate's WHOLE span, which for the fat sides this path serves
689/// would prefetch the very extent the probe strategy exists to avoid.
690pub(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
712/// Batch-fault the tiles that probing `patterns` from `subject_ids` (each
713/// bound onto the subject variable `sv`) will route to — one coalesced read
714/// per section instead of one round trip per candidate. Purely a cache
715/// warmer: correctness is untouched, and a local index is a no-op. Used by
716/// the FILTER-CONTAINS pushdown before it probes its candidate subjects.
717pub(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
754/// A pattern whose own constant terms (a bound subject or object) already route
755/// it to a narrow index range — a few coalesced range reads to scan in full.
756fn pattern_is_selective(t: &(SlotTerm, SlotTerm, SlotTerm)) -> bool {
757    matches!(t.0, SlotTerm::Node(_)) || matches!(t.2, SlotTerm::Node(_))
758}
759
760/// The (deduped) slots a lowered pattern binds.
761fn 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
773/// The row value of a matched triple's predicate `p_id`, for a pattern whose
774/// predicate slot is `slot`. Only a variable predicate binds it, so only then
775/// is it canonicalized (a predicate that is also a node becomes the node id):
776/// that is a dictionary lookup of the predicate as a node, memoized per query,
777/// but up to three dependent chunk reads on a lazy open for each new predicate
778/// — reads a constant predicate never needed. Otherwise the value is unused.
779fn 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
786/// Build a solution row for one scanned triple, enforcing repeated variables
787/// *within* the pattern (e.g. `?x p ?x`). `None` = the triple doesn't satisfy
788/// a repeated variable.
789fn triple_row(
790    ctx: &Ctx,
791    t: &(SlotTerm, SlotTerm, SlotTerm),
792    (s_id, p_id, o_id): (u32, u32, u32),
793) -> Option<Row> {
794    // A query under a memory ceiling ([`crate::guard`]) stops between rows.
795    crate::guard::check();
796    let dict = ctx.rete.dictionary();
797    let s_val = dict.subject_node(s_id) as i64;
798    let p_val = pred_val(ctx, &t.1, p_id);
799    let o_val = dict.object_node(o_id) as i64;
800    let mut row = ctx.slots.empty_row();
801    for (term, val) in [(&t.0, s_val), (&t.1, p_val), (&t.2, o_val)] {
802        if let SlotTerm::Var(i) = term {
803            match row[*i] {
804                Some(Val::Id(existing)) if existing != val => return None,
805                Some(_) => {}
806                None => row[*i] = Some(Val::Id(val)),
807            }
808        }
809    }
810    Some(row)
811}
812
813/// Lazily scan one lowered pattern as a stream of solution rows. The scan
814/// constrains only the pattern's constant terms (bound variables are joined in
815/// afterwards, not pushed into the scan). `None` = a constant is unsatisfiable
816/// (unknown to the dictionary or in an impossible role), which empties the BGP.
817fn scan_rows<'q>(
818    ctx: &'q Ctx<'q>,
819    index: &'q GraphIndex,
820    t: (SlotTerm, SlotTerm, SlotTerm),
821) -> Option<impl Iterator<Item = Row> + 'q> {
822    let dict = ctx.rete.dictionary();
823    let (sid, pid, oid) = (
824        const_subject(&t.0, dict)?,
825        const_predicate(&t.1)?,
826        const_object(&t.2, dict)?,
827    );
828    Some(
829        index
830            .scan_iter((sid, pid, oid))
831            .filter_map(move |triple| triple_row(ctx, &t, triple)),
832    )
833}
834
835/// `GRAPH ?g { one pattern }` over a quad-index file: every match of the
836/// pattern in any graph as a row that also binds slot `gslot` to the graph,
837/// streamed from one pass over the tiles (no view per graph). `None` when this
838/// does not apply (not a single pattern, the graph variable also inside the
839/// pattern, a file without a quad index): the caller falls back to walking the
840/// graphs. A constant the dictionary cannot place gives an empty stream.
841pub(crate) fn quad_scan_rows<'q>(
842    ctx: &'q Ctx<'q>,
843    patterns: &[TriplePattern],
844    gslot: usize,
845) -> Option<Box<dyn Iterator<Item = Row> + 'q>> {
846    let [p] = patterns else {
847        return None;
848    };
849    if [&p.s, &p.p, &p.o]
850        .iter()
851        .any(|t| matches!(t, PatternTerm::Var(v) if ctx.slots.slot(v) == Some(gslot)))
852    {
853        return None;
854    }
855    let dict = ctx.rete.dictionary();
856    let Some(lowered) = lower(patterns, ctx) else {
857        return Some(Box::new(std::iter::empty()));
858    };
859    let t = lowered[0];
860    let (Some(sid), Some(pid), Some(oid)) = (
861        const_subject(&t.0, dict),
862        const_predicate(&t.1),
863        const_object(&t.2, dict),
864    ) else {
865        return Some(Box::new(std::iter::empty()));
866    };
867    let scan = ctx.rete.named_quad_scan((sid, pid, oid))?;
868    Some(Box::new(scan.filter_map(move |(g, triple)| {
869        let mut row = triple_row(ctx, &t, triple)?;
870        row[gslot] = Some(Val::Id(dict.subject_node(g) as i64));
871        Some(row)
872    })))
873}
874
875/// A triple pattern as the dictionary ids of its constants (subject, predicate, object).
876pub(crate) type ScanPattern = (Option<u32>, Option<u32>, Option<u32>);
877
878/// The pattern of a BGP that best narrows a scan over every graph, as the
879/// dictionary ids of its constants: the one with the most constants, a bound
880/// subject or object weighing more than a bound predicate. `None` when no
881/// pattern has a constant, or one cannot be placed in the dictionary.
882pub(crate) fn driver_pattern(ctx: &Ctx, patterns: &[TriplePattern]) -> Option<ScanPattern> {
883    let dict = ctx.rete.dictionary();
884    let lowered = lower(patterns, ctx)?;
885    let mut best: Option<(u32, ScanPattern)> = None;
886    for t in &lowered {
887        let (sid, pid, oid) = (
888            const_subject(&t.0, dict)?,
889            const_predicate(&t.1)?,
890            const_object(&t.2, dict)?,
891        );
892        let score =
893            3 * u32::from(sid.is_some()) + u32::from(pid.is_some()) + 3 * u32::from(oid.is_some());
894        if score > 0 && best.is_none_or(|(b, _)| score > b) {
895            best = Some((score, (sid, pid, oid)));
896        }
897    }
898    best.map(|(_, pat)| pat)
899}
900
901/// Scan one lowered pattern into a materialized relation, returning the rows
902/// and the slots the pattern binds. `None` = a constant is unsatisfiable.
903fn pattern_rows(
904    ctx: &Ctx,
905    index: &GraphIndex,
906    t: &(SlotTerm, SlotTerm, SlotTerm),
907) -> Option<(Vec<Row>, Vec<usize>)> {
908    let slots = pattern_slots(t);
909    let mut rel: Vec<Row> = Vec::new();
910    let dict = ctx.rete.dictionary();
911    let (sid, pid, oid) = (
912        const_subject(&t.0, dict)?,
913        const_predicate(&t.1)?,
914        const_object(&t.2, dict)?,
915    );
916    // Stream the matches with the lazy cursor — the hash join is
917    // order-independent, so no canonical re-sort is needed here.
918    for triple in index.scan_iter((sid, pid, oid)) {
919        if let Some(row) = triple_row(ctx, t, triple) {
920            rel.push(row);
921        }
922    }
923    Some((rel, slots))
924}
925
926/// A per-pattern cardinality estimate, smaller = cheaper to scan. The base is
927/// the **exact** per-predicate triple count (sum of the summary quotient graph's
928/// super-edge counts) when the predicate is bound, else the whole-graph total.
929/// A bound subject/object (a constant, or a variable already bound by `seed`)
930/// then scales it down by that predicate's **measured selectivity** from the
931/// `query_stats` block — `1 / distinct_subjects` for a bound subject (so
932/// `<s> <p> ?o` ≈ the average objects per subject; exactly 1 for a functional
933/// predicate), `1 / distinct_objects` for a bound object. Files built before the
934/// `query_stats` block fall back to fixed default selectivities. `None` when the
935/// file carries no pyramid summary at all — ordering then uses the constant-count
936/// heuristic.
937fn pattern_estimates(
938    ctx: &Ctx,
939    index: &GraphIndex,
940    lowered: &[(SlotTerm, SlotTerm, SlotTerm)],
941    seed: &std::collections::HashSet<usize>,
942) -> Option<Vec<f64>> {
943    // Only use the summary when it's already in memory — never fault it just to
944    // plan (the lazy remote path defers the pyramid by design).
945    let pyr = ctx.rete.pyramid_if_loaded()?;
946    let mut pred: std::collections::HashMap<u32, u64> = std::collections::HashMap::new();
947    for e in &pyr.summary {
948        *pred.entry(e.predicate).or_insert(0) += e.count as u64;
949    }
950    // Measured per-predicate distinct subjects/objects: the SCHEMA section
951    // (kind 9) — where a file built since it existed keeps them, its
952    // pyramid-meta holding only the community summary — else the pyramid-meta's
953    // query-stats block; empty on files built before either → the default
954    // selectivities. The pyramid is resident here, so the (small) section is
955    // too on a resident open, and costs one range once on a lazy one.
956    let stats: std::collections::HashMap<u32, &crate::meta::PredStat> = ctx
957        .rete
958        .planner_stats(true)
959        .iter()
960        .map(|s| (s.predicate, s))
961        .collect();
962    // The active graph's own size: a pattern inside `GRAPH <g>` cannot match more
963    // triples than `g` holds, whatever the file-wide predicate count says.
964    let graph_total = ctx.rete.index_triples(index).max(1) as f64;
965    let total = graph_total;
966    let num_preds = pred.len().max(1) as f64;
967    // Defaults used only when query_stats has no entry for the predicate.
968    const SEL_SUBJECT: f64 = 0.001;
969    const SEL_OBJECT: f64 = 0.02;
970    let node_bound = |t: &SlotTerm| match t {
971        SlotTerm::Node(_) => true,
972        SlotTerm::Var(v) => seed.contains(v),
973        SlotTerm::Pred(_) => false,
974    };
975    Some(
976        lowered
977            .iter()
978            .map(|t| {
979                // Base from the predicate: exact total for a constant predicate,
980                // the average predicate total for a seed-bound predicate variable
981                // (the value is known to the probe but not at plan time), else the
982                // whole graph. `st` is the measured stats for a constant predicate.
983                let (base, st) = match t.1 {
984                    SlotTerm::Pred(p) => {
985                        (*pred.get(&p).unwrap_or(&0) as f64, stats.get(&p).copied())
986                    }
987                    SlotTerm::Var(v) if seed.contains(&v) => (total / num_preds, None),
988                    _ => (total, None),
989                };
990                let mut est = base.min(graph_total).max(1.0);
991                if node_bound(&t.0) {
992                    est *= match st {
993                        Some(s) if s.distinct_subjects > 0 => 1.0 / s.distinct_subjects as f64,
994                        _ => SEL_SUBJECT,
995                    };
996                }
997                if node_bound(&t.2) {
998                    est *= match st {
999                        Some(s) if s.distinct_objects > 0 => 1.0 / s.distinct_objects as f64,
1000                        _ => SEL_OBJECT,
1001                    };
1002                }
1003                est.max(1.0)
1004            })
1005            .collect(),
1006    )
1007}
1008
1009/// The minimum, over a BGP's patterns, of [`pattern_scan_bytes`] — the
1010/// cheapest single-pattern scan the hash-join path would at least perform to
1011/// materialize this BGP. `None` = a constant term is unknown to the
1012/// dictionary (the BGP is unsatisfiable; the caller short-circuits anyway).
1013/// Unlike [`pattern_estimates`], this needs no pyramid summary, so it works on
1014/// remote-lazy files — exactly where the answer matters most.
1015pub(crate) fn bgp_min_scan_bytes(
1016    ctx: &Ctx,
1017    index: &GraphIndex,
1018    patterns: &[TriplePattern],
1019) -> Option<u64> {
1020    let lowered = lower(patterns, ctx)?;
1021    lowered
1022        .iter()
1023        .map(|t| pattern_scan_bytes(ctx, index, t))
1024        .min()
1025}
1026
1027/// One pattern's cheapest scan extent in BYTES — what `rete cost` previews
1028/// (the planner's gates use row estimates instead): for each permutation, sum the
1029/// encoded lengths of the tiles a scan of this pattern would route to (all
1030/// tiles covering the bound leading key, or the whole section when the leading
1031/// key is unbound), and take the minimum. Tile *counts* are useless here — a
1032/// broad predicate can sit inside one giant tile — but the directory's byte
1033/// lengths expose it. Reads only resident metadata, never tile data.
1034///
1035/// A bound term routes by its id **in the role its position has**: a tile's
1036/// leading range is in subject-, predicate- or object-id space, not in the
1037/// unified node space a lowered constant carries. Routing the node id instead
1038/// was off by one for subjects and off by the whole subject-only block for an
1039/// object-only term (every literal), so `?id :value "10.1234/x"` could cost
1040/// "0 bytes" and pass the merge-join gate against a fat partner, which the
1041/// merge then materialized in full. A constant that has no id in the role
1042/// (never a subject, say) matches nothing: 0 bytes is then exact.
1043fn pattern_scan_bytes(ctx: &Ctx, index: &GraphIndex, t: &(SlotTerm, SlotTerm, SlotTerm)) -> u64 {
1044    let dict = ctx.rete.dictionary();
1045    match (
1046        const_subject(&t.0, dict),
1047        const_predicate(&t.1),
1048        const_object(&t.2, dict),
1049    ) {
1050        (Some(s), Some(p), Some(o)) => routed_scan_bytes(index, (s, p, o)),
1051        _ => 0,
1052    }
1053}
1054
1055/// [`pattern_scan_bytes`] for a pattern already in role-id space.
1056pub(crate) fn routed_scan_bytes(index: &GraphIndex, pattern: Pattern) -> u64 {
1057    let sections = index.tile_sections();
1058    let mut best = u64::MAX;
1059    // Only the permutations the FILE carries: an absent section is an empty
1060    // `Vec<Tile>`, and summing its (zero) tile lengths would report a free scan
1061    // for the one permutation that cannot be scanned at all.
1062    for perm in index.perms().iter() {
1063        let tiles = sections[perm.section_index()];
1064        // A leafed (0x06) directory answers from its root without faulting a
1065        // leaf: a cost estimate must not cost a round trip.
1066        best = best.min(match perm.order_pattern(pattern)[0] {
1067            Some(id) => tiles.routed_bytes(id),
1068            None => tiles.total_bytes(),
1069        });
1070    }
1071    best
1072}
1073
1074/// Estimated matches above which a join side counts as "fat": materializing
1075/// (or, on a remote index, fetching) a million rows to build a hash/merge side
1076/// is past the point where probing it per surviving row wins. Shared by the
1077/// merge-seed asymmetry gate here and the force-probe gates in `sparql::eval`.
1078///
1079/// A ROW estimate ([`GraphIndex::estimate_matches`]: routed tiles × rows per
1080/// tile), not a byte figure. Tiles are cut by their encoded size before
1081/// compression, so the estimate does not move when only the compression does;
1082/// the compressed tile bytes this gate used to read did — at zstd level 19,
1083/// two OpenCitations queries crossed the old 2 MiB line and ran 3–7× slower
1084/// with up to 6× the memory — and they differed between a resident open
1085/// (decoded sizes) and a lazy one (compressed sizes) of the same file.
1086pub(crate) const FAT_SCAN_ROWS: f64 = (1u64 << 20) as f64;
1087
1088/// Would a merge join of `ta` and `tb` materialize a fat side next to a
1089/// pinpoint one? (One side above [`FAT_SCAN_ROWS`] and at least four times the
1090/// other.) Then the hybrid loop — scan the small side, probe the fat one — is
1091/// strictly better, and the merge seed is declined.
1092fn merge_seed_lopsided(
1093    ctx: &Ctx,
1094    index: &GraphIndex,
1095    ta: &(SlotTerm, SlotTerm, SlotTerm),
1096    tb: &(SlotTerm, SlotTerm, SlotTerm),
1097) -> bool {
1098    let est = tile_estimates(ctx, index, &[*ta, *tb], &std::collections::HashSet::new());
1099    let (lo, hi) = (est[0].min(est[1]), est[0].max(est[1]));
1100    hi >= FAT_SCAN_ROWS && hi / 4.0 >= lo
1101}
1102
1103/// Order pattern indices for a left-deep join: cheapest (smallest estimated
1104/// cardinality) first, always preferring a pattern that shares a variable with
1105/// the already-joined set so the join stays connected and intermediate relations
1106/// stay small. Cardinality comes from [`pattern_estimates`]; with no summary it
1107/// falls back to a most-constants-first heuristic. Pure reordering — `hash_join`
1108/// is order-independent, so the result multiset is unchanged.
1109fn selectivity_order(
1110    ctx: &Ctx,
1111    index: &GraphIndex,
1112    lowered: &[(SlotTerm, SlotTerm, SlotTerm)],
1113) -> Vec<usize> {
1114    selectivity_order_seeded(ctx, index, lowered, &std::collections::HashSet::new())
1115}
1116
1117/// [`selectivity_order`] with a set of slots already bound by an outer seed
1118/// row: those variables count as already connected (and, in the no-summary
1119/// fallback, as constants for selectivity).
1120fn selectivity_order_seeded(
1121    ctx: &Ctx,
1122    index: &GraphIndex,
1123    lowered: &[(SlotTerm, SlotTerm, SlotTerm)],
1124    seed: &std::collections::HashSet<usize>,
1125) -> Vec<usize> {
1126    // Exact per-predicate statistics when the pyramid is resident; otherwise
1127    // (the lazy path, and files built without a pyramid) the tile-directory
1128    // estimate, which needs no fetch. That one also counts a class's own
1129    // instances (`?s a :C` routes to the tiles of `:C`), so the
1130    // class-enumeration rule below, a stand-in for exactly that missing
1131    // number, applies only to the pyramid figures, whose `rdf:type`
1132    // selectivity is an average over classes.
1133    let (estimates, per_class) = match pattern_estimates(ctx, index, lowered, seed) {
1134        Some(e) => (Some(e), false),
1135        None => (Some(tile_estimates(ctx, index, lowered, seed)), true),
1136    };
1137    let consts = |t: &(SlotTerm, SlotTerm, SlotTerm)| {
1138        [&t.0, &t.1, &t.2]
1139            .into_iter()
1140            .filter(|x| match x {
1141                SlotTerm::Var(v) => seed.contains(v),
1142                _ => true,
1143            })
1144            .count()
1145    };
1146    let vars = |t: &(SlotTerm, SlotTerm, SlotTerm)| -> Vec<usize> {
1147        [&t.0, &t.1, &t.2]
1148            .into_iter()
1149            .filter_map(|x| match x {
1150                SlotTerm::Var(v) => Some(*v),
1151                _ => None,
1152            })
1153            .collect()
1154    };
1155    // Higher score = picked sooner. With summary stats that's the negated
1156    // estimate (smaller cardinality wins); without, the constant count (old
1157    // behaviour, byte-identical ordering for pyramid-less files).
1158    let score = |i: usize| -> f64 {
1159        match &estimates {
1160            Some(e) => -e[i],
1161            None => consts(&lowered[i]) as f64,
1162        }
1163    };
1164    // A `?s rdf:type <Class>` with an as-yet-unbound subject is a **class
1165    // enumeration**: it matches every instance of the class, so it is almost never
1166    // the most selective place to *start* a join (a popular class has thousands of
1167    // instances), even though it has two bound positions and so looks selective to
1168    // the constant-count heuristic. Deprioritize it as a seed; once its subject is
1169    // bound by another pattern it becomes a cheap per-row type *check* (the hybrid
1170    // join probes it), so pushing it later costs nothing. Robust and stats-free.
1171    let type_pid = ctx.rete.dictionary().predicate_id(crate::file::RDF_TYPE);
1172    let is_class_enum = |i: usize, bound: &std::collections::HashSet<usize>| -> bool {
1173        let t = &lowered[i];
1174        !per_class
1175            && type_pid.is_some_and(|tp| matches!(t.1, SlotTerm::Pred(p) if p == tp))
1176            && matches!(t.2, SlotTerm::Node(_))
1177            && matches!(t.0, SlotTerm::Var(v) if !bound.contains(&v))
1178    };
1179    let n = lowered.len();
1180    let mut remaining: Vec<usize> = (0..n).collect();
1181    let mut order: Vec<usize> = Vec::with_capacity(n);
1182    let mut bound: std::collections::HashSet<usize> = seed.clone();
1183    while !remaining.is_empty() {
1184        // Pick the remaining pattern with the best (connected, not-a-class-enum,
1185        // score) key; ties go to the lowest original index for a stable order.
1186        let best = *remaining
1187            .iter()
1188            .max_by(|&&a, &&b| {
1189                let connected = |i: usize| vars(&lowered[i]).iter().any(|v| bound.contains(v));
1190                connected(a)
1191                    .cmp(&connected(b))
1192                    .then_with(|| is_class_enum(b, &bound).cmp(&is_class_enum(a, &bound)))
1193                    .then_with(|| {
1194                        score(a)
1195                            .partial_cmp(&score(b))
1196                            .unwrap_or(std::cmp::Ordering::Equal)
1197                    })
1198                    .then_with(|| b.cmp(&a))
1199            })
1200            .unwrap();
1201        for v in vars(&lowered[best]) {
1202            bound.insert(v);
1203        }
1204        order.push(best);
1205        remaining.retain(|&i| i != best);
1206    }
1207    order
1208}
1209
1210/// Does the BGP have at least one solution against `index`? A single pattern
1211/// with all-distinct variables streams the index and stops at the first match
1212/// (no materialization); anything else falls back to the full evaluator and
1213/// tests non-emptiness (still benefiting from the lazy per-pattern scan). Used
1214/// by `ASK`.
1215pub(crate) fn bgp_exists(ctx: &Ctx, index: &GraphIndex, patterns: &[TriplePattern]) -> bool {
1216    let dict = ctx.rete.dictionary();
1217    let Some(lowered) = lower(patterns, ctx) else {
1218        return false;
1219    };
1220    if let [t] = lowered.as_slice() {
1221        // The fast path can't enforce a variable repeated across positions
1222        // (e.g. `?x p ?x`) — that needs the row builder — so only take it when
1223        // the pattern's variables are all distinct.
1224        let names: Vec<usize> = [&t.0, &t.1, &t.2]
1225            .into_iter()
1226            .filter_map(|x| match x {
1227                SlotTerm::Var(v) => Some(*v),
1228                _ => None,
1229            })
1230            .collect();
1231        let distinct = names
1232            .iter()
1233            .enumerate()
1234            .all(|(i, v)| !names[i + 1..].contains(v));
1235        if distinct {
1236            return match (
1237                const_subject(&t.0, dict),
1238                const_predicate(&t.1),
1239                const_object(&t.2, dict),
1240            ) {
1241                (Some(s), Some(p), Some(o)) => index.scan_iter((s, p, o)).next().is_some(),
1242                _ => false,
1243            };
1244        }
1245    }
1246    !eval_bgp_rows(ctx, index, patterns).is_empty()
1247}
1248
1249/// A lazy left-deep BGP join that yields solution rows one at a time, so a
1250/// consumer under `LIMIT`/`OFFSET` (or `.next()` for ASK) can stop early —
1251/// stopping also stops the underlying index scan.
1252///
1253/// The all-but-last patterns are joined eagerly into a `prefix` **hash table**
1254/// (they are the most selective, so this is the small side); the *last* (least
1255/// selective) pattern is then **streamed** from the index cursor, probing the
1256/// prefix — the big scan is never materialized, and a `LIMIT` above stops it
1257/// after a handful of triples. Yields the same solution multiset as
1258/// [`eval_bgp_rows`] (the join is order-independent), just incrementally.
1259pub(crate) struct BgpSolutions<'q> {
1260    scan: Option<Box<dyn Iterator<Item = Row> + 'q>>,
1261    prefix: Vec<Row>,
1262    /// Prefix row indices keyed by the shared-slot values.
1263    buckets: HashMap<Vec<Val>, Vec<usize>>,
1264    shared: Vec<usize>,
1265    cartesian: bool,
1266    /// The prefix is the all-unbound seed row (single-pattern BGP): scanned
1267    /// rows are already complete solutions — pass them through unmerged.
1268    seed_only: bool,
1269    cur_scan: Option<Row>,
1270    /// Reused candidate buffer (avoids a Vec allocation per scanned row).
1271    matches: Vec<usize>,
1272    mi: usize,
1273}
1274
1275impl<'q> BgpSolutions<'q> {
1276    /// Yield `rows` as complete solutions (no prefix merge).
1277    fn passthrough(rows: Box<dyn Iterator<Item = Row> + 'q>) -> Self {
1278        BgpSolutions {
1279            scan: Some(rows),
1280            prefix: Vec::new(),
1281            buckets: HashMap::new(),
1282            shared: Vec::new(),
1283            cartesian: false,
1284            seed_only: true,
1285            cur_scan: None,
1286            matches: Vec::new(),
1287            mi: 0,
1288        }
1289    }
1290
1291    /// The fully streaming join, for a BGP whose most selective pattern is
1292    /// itself large ([`STREAM_SEED_ROWS`]). Nothing proportional to it is ever
1293    /// held: the seed pattern streams from its index cursor (or, when the next
1294    /// pattern is just as large and shares its subject variable, both stream
1295    /// through a merge join), and each later pattern joins per row — through
1296    /// a hash table when the pattern is small ([`HASH_BUILD_ROWS`]), by an
1297    /// index probe otherwise. Same solution multiset as the materializing
1298    /// path; only the row order differs.
1299    fn streaming(
1300        ctx: &'q Ctx<'q>,
1301        index: &'q GraphIndex,
1302        lowered: &[(SlotTerm, SlotTerm, SlotTerm)],
1303        order: &[usize],
1304        est: &[f64],
1305    ) -> Self {
1306        let (mut stream, mut bound, skip): (Box<dyn Iterator<Item = Row> + 'q>, Vec<usize>, usize) =
1307            match (est[order[1]] > limits().stream_seed)
1308                .then(|| merge_stream(ctx, index, lowered[order[0]], lowered[order[1]]))
1309                .flatten()
1310            {
1311                Some((m, slots)) => (m, slots, 2),
1312                None => {
1313                    let t = lowered[order[0]];
1314                    let Some(scan) = scan_rows(ctx, index, t) else {
1315                        return Self::empty();
1316                    };
1317                    (Box::new(scan), pattern_slots(&t), 1)
1318                }
1319            };
1320        for &i in &order[skip..] {
1321            let t = lowered[i];
1322            let slots = pattern_slots(&t);
1323            let shared: Vec<usize> = slots
1324                .iter()
1325                .copied()
1326                .filter(|s| bound.contains(s))
1327                .collect();
1328            if !shared.is_empty() && est[i] > limits().hash_build {
1329                stream = Box::new(stream.flat_map(move |base| probe_rows(ctx, index, t, base)));
1330            } else {
1331                // Small (or unconnected) pattern: scan it once into a table
1332                // keyed on the shared slots and join every streamed row to it.
1333                let Some((rel, _)) = pattern_rows(ctx, index, &t) else {
1334                    return Self::empty();
1335                };
1336                let mut table: HashMap<Vec<Val>, Vec<Row>> = HashMap::new();
1337                for r in rel {
1338                    let key: Vec<Val> = shared.iter().map(|&s| r[s].clone().unwrap()).collect();
1339                    table.entry(key).or_default().push(r);
1340                }
1341                let fill = slots.clone();
1342                stream = Box::new(stream.flat_map(move |base| {
1343                    let key: Vec<Val> = shared.iter().map(|&s| base[s].clone().unwrap()).collect();
1344                    let out: Vec<Row> = table
1345                        .get(&key)
1346                        .map(|rs| {
1347                            rs.iter()
1348                                .map(|r| {
1349                                    let mut row = base.clone();
1350                                    for &s in &fill {
1351                                        row[s] = r[s].clone();
1352                                    }
1353                                    row
1354                                })
1355                                .collect()
1356                        })
1357                        .unwrap_or_default();
1358                    out.into_iter()
1359                }));
1360            }
1361            for s in slots {
1362                if !bound.contains(&s) {
1363                    bound.push(s);
1364                }
1365            }
1366        }
1367        Self::passthrough(stream)
1368    }
1369
1370    /// An iterator that yields nothing (an unsatisfiable BGP).
1371    fn empty() -> Self {
1372        BgpSolutions {
1373            scan: None,
1374            prefix: Vec::new(),
1375            buckets: HashMap::new(),
1376            shared: Vec::new(),
1377            cartesian: false,
1378            seed_only: false,
1379            cur_scan: None,
1380            matches: Vec::new(),
1381            mi: 0,
1382        }
1383    }
1384
1385    pub(crate) fn new(ctx: &'q Ctx<'q>, index: &'q GraphIndex, patterns: &[TriplePattern]) -> Self {
1386        // An empty BGP has exactly one (empty) solution.
1387        if patterns.is_empty() {
1388            return BgpSolutions {
1389                scan: Some(Box::new(std::iter::once(ctx.slots.empty_row()))),
1390                prefix: vec![ctx.slots.empty_row()],
1391                buckets: HashMap::new(),
1392                shared: Vec::new(),
1393                cartesian: true,
1394                seed_only: true,
1395                cur_scan: None,
1396                matches: Vec::new(),
1397                mi: 0,
1398            };
1399        }
1400        // Lower all patterns; an unknown constant term ⇒ no solutions.
1401        let Some(lowered) = lower(patterns, ctx) else {
1402            return Self::empty();
1403        };
1404        // Join all but the *last* (least selective) pattern eagerly into the
1405        // prefix; a single pattern leaves the seed row as the prefix.
1406        let order = selectivity_order(ctx, index, &lowered);
1407        let scan_est = tile_estimates(ctx, index, &lowered, &std::collections::HashSet::new());
1408        // Even the most selective pattern is big: materializing it as the
1409        // prefix would hold every one of its rows. Stream the whole join.
1410        if order.len() >= 2 && scan_est[order[0]] > limits().stream_seed {
1411            return Self::streaming(ctx, index, &lowered, &order, &scan_est);
1412        }
1413        let (&last_i, prefix_is) = order.split_last().unwrap();
1414        let prefix_pats: Vec<TriplePattern> =
1415            prefix_is.iter().map(|&i| patterns[i].clone()).collect();
1416        let prefix = eval_bgp_rows(ctx, index, &prefix_pats);
1417        if prefix.is_empty() {
1418            return Self::empty();
1419        }
1420
1421        // Hash-join key = slots shared by the prefix and the last pattern (BGP
1422        // rows bind every slot of their pattern set, so the prefix's bound set
1423        // can be read off its first row).
1424        let shared: Vec<usize> = pattern_slots(&lowered[last_i])
1425            .into_iter()
1426            .filter(|&s| prefix[0][s].is_some())
1427            .collect();
1428
1429        // A few prefix rows against a big last pattern: look each one up
1430        // instead of scanning the pattern. `?r :relationSubject <doi> . ?r ?p
1431        // ?o` has four prefix rows and a last pattern that is the whole graph.
1432        // The last pattern is an unbound class (`?x rdf:type ?c`) whose subject
1433        // the prefix binds: each row's types are one SPO lookup, and the prefix
1434        // is already resident, so probing costs no memory the scan would not.
1435        // It wins until the rows outnumber the statements the scan would read
1436        // (the usual sixteenth gives way to one).
1437        let class_probe = !shared.is_empty()
1438            && scan_est[last_i] >= limits().probe_floor
1439            && (prefix.len() as f64) <= scan_est[last_i]
1440            && is_unbound_class(ctx, &lowered[last_i], &shared);
1441        if !shared.is_empty() && (probe_beats_scan(prefix.len(), scan_est[last_i]) || class_probe) {
1442            let t = lowered[last_i];
1443            if index.is_remote() {
1444                prefetch_rows_for(ctx, index, t, &prefix);
1445            }
1446            let stream = prefix
1447                .into_iter()
1448                .flat_map(move |base| probe_rows(ctx, index, t, base));
1449            return Self::passthrough(Box::new(stream));
1450        }
1451
1452        // Stream the last pattern with the lazy index cursor.
1453        let Some(scan) = scan_rows(ctx, index, lowered[last_i]) else {
1454            return Self::empty();
1455        };
1456        let cartesian = shared.is_empty();
1457        // A single-pattern BGP joins against the all-unbound seed: scanned rows
1458        // are already complete solutions.
1459        let seed_only = prefix.len() == 1 && prefix[0].iter().all(Option::is_none);
1460        let mut buckets: HashMap<Vec<Val>, Vec<usize>> = HashMap::new();
1461        if !cartesian {
1462            for (i, r) in prefix.iter().enumerate() {
1463                let key: Vec<Val> = shared.iter().map(|&s| r[s].clone().unwrap()).collect();
1464                buckets.entry(key).or_default().push(i);
1465            }
1466        }
1467        BgpSolutions {
1468            scan: Some(Box::new(scan)),
1469            prefix,
1470            buckets,
1471            shared,
1472            cartesian,
1473            seed_only,
1474            cur_scan: None,
1475            matches: Vec::new(),
1476            mi: 0,
1477        }
1478    }
1479}
1480
1481impl Iterator for BgpSolutions<'_> {
1482    type Item = Row;
1483
1484    fn next(&mut self) -> Option<Row> {
1485        // Single-pattern fast path: pass scanned rows through unmerged.
1486        if self.seed_only {
1487            return self.scan.as_mut()?.next();
1488        }
1489        loop {
1490            // Emit the next prefix match for the current scanned row.
1491            if self.mi < self.matches.len() {
1492                let pi = self.matches[self.mi];
1493                self.mi += 1;
1494                let mut merged = self.prefix[pi].clone();
1495                for (slot, v) in self.cur_scan.as_ref().unwrap().iter().enumerate() {
1496                    if v.is_some() {
1497                        merged[slot] = v.clone();
1498                    }
1499                }
1500                return Some(merged);
1501            }
1502            // Pull the next row from the index scan and gather its matches
1503            // (into the reused buffer — no allocation per scanned row).
1504            let s = self.scan.as_mut()?.next()?;
1505            self.matches.clear();
1506            if self.cartesian {
1507                self.matches.extend(0..self.prefix.len());
1508            } else {
1509                let key: Vec<Val> = self.shared.iter().map(|&i| s[i].clone().unwrap()).collect();
1510                if let Some(c) = self.buckets.get(&key) {
1511                    self.matches.extend_from_slice(c);
1512                }
1513            }
1514            self.mi = 0;
1515            self.cur_scan = Some(s);
1516        }
1517    }
1518}
1519
1520/// Batch-fault the tiles that probing `t` from each of `rows` will route to
1521/// (remote readers only; see [`GraphIndex::prefetch_probe_tiles`]).
1522fn prefetch_rows_for(
1523    ctx: &Ctx,
1524    index: &GraphIndex,
1525    t: (SlotTerm, SlotTerm, SlotTerm),
1526    rows: &[Row],
1527) {
1528    let pats: Vec<Pattern> = rows
1529        .iter()
1530        .filter_map(|base| {
1531            match (
1532                probe_subject(ctx, &t.0, base),
1533                probe_predicate(ctx, &t.1, base),
1534                probe_object(ctx, &t.2, base),
1535            ) {
1536                (Some(s), Some(p), Some(o)) => Some((s, p, o)),
1537                _ => None,
1538            }
1539        })
1540        .collect();
1541    index.prefetch_probe_tiles(&pats);
1542}
1543
1544/// A **streaming** merge join of two patterns that share exactly one variable,
1545/// in the subject position of both: each side streams from a permutation
1546/// sorted on subject ([`GraphIndex::scan_iter_sorted_on`]) and only the
1547/// current subject's rows are held. Restricted to the subject column because a
1548/// subject's rows are its out-edges, a small run; an object run (every
1549/// instance of a class) can be the whole pattern. `None` when the shape does
1550/// not qualify; the caller then streams the first pattern alone.
1551#[allow(clippy::type_complexity)]
1552fn merge_stream<'q>(
1553    ctx: &'q Ctx<'q>,
1554    index: &'q GraphIndex,
1555    ta: (SlotTerm, SlotTerm, SlotTerm),
1556    tb: (SlotTerm, SlotTerm, SlotTerm),
1557) -> Option<(Box<dyn Iterator<Item = Row> + 'q>, Vec<usize>)> {
1558    let sa = pattern_slots(&ta);
1559    let sb = pattern_slots(&tb);
1560    let shared: Vec<usize> = sa.iter().copied().filter(|s| sb.contains(s)).collect();
1561    let [v] = shared[..] else {
1562        return None;
1563    };
1564    if !matches!(ta.0, SlotTerm::Var(i) if i == v) || !matches!(tb.0, SlotTerm::Var(i) if i == v) {
1565        return None;
1566    }
1567    let dict = ctx.rete.dictionary();
1568    let lower = |t: &(SlotTerm, SlotTerm, SlotTerm)| -> Option<Pattern> {
1569        Some((
1570            const_subject(&t.0, dict)?,
1571            const_predicate(&t.1)?,
1572            const_object(&t.2, dict)?,
1573        ))
1574    };
1575    let side = |t: (SlotTerm, SlotTerm, SlotTerm)| -> Option<KeyedRows<'q>> {
1576        let it = index.scan_iter_sorted_on(lower(&t)?, 0)?;
1577        let keyed: Box<dyn Iterator<Item = (u32, Row)> + 'q> =
1578            Box::new(it.filter_map(move |tri| triple_row(ctx, &t, tri).map(|r| (tri.0, r))));
1579        Some(keyed.peekable())
1580    };
1581    let a = side(ta)?;
1582    let b = side(tb)?;
1583    let mut slots = sa;
1584    for s in sb {
1585        if !slots.contains(&s) {
1586            slots.push(s);
1587        }
1588    }
1589    Some((
1590        Box::new(MergeStream {
1591            a,
1592            b,
1593            ra: Vec::new(),
1594            rb: Vec::new(),
1595            i: 0,
1596            j: 0,
1597        }),
1598        slots,
1599    ))
1600}
1601
1602type KeyedRows<'q> = std::iter::Peekable<Box<dyn Iterator<Item = (u32, Row)> + 'q>>;
1603
1604/// The two-pointer merge behind [`merge_stream`]: `ra`/`rb` hold the current
1605/// equal-key runs and `(i, j)` walks their cross product lazily.
1606struct MergeStream<'q> {
1607    a: KeyedRows<'q>,
1608    b: KeyedRows<'q>,
1609    ra: Vec<Row>,
1610    rb: Vec<Row>,
1611    i: usize,
1612    j: usize,
1613}
1614
1615impl Iterator for MergeStream<'_> {
1616    type Item = Row;
1617
1618    fn next(&mut self) -> Option<Row> {
1619        loop {
1620            while self.i < self.ra.len() {
1621                if self.j < self.rb.len() {
1622                    let j = self.j;
1623                    self.j += 1;
1624                    if let Some(m) = crate::row::merge_rows(&self.ra[self.i], &self.rb[j]) {
1625                        return Some(m);
1626                    }
1627                } else {
1628                    self.i += 1;
1629                    self.j = 0;
1630                }
1631            }
1632            let ka = self.a.peek()?.0;
1633            let kb = self.b.peek()?.0;
1634            match ka.cmp(&kb) {
1635                std::cmp::Ordering::Less => {
1636                    self.a.next();
1637                }
1638                std::cmp::Ordering::Greater => {
1639                    self.b.next();
1640                }
1641                std::cmp::Ordering::Equal => {
1642                    self.ra.clear();
1643                    self.rb.clear();
1644                    while let Some((_, r)) = self.a.next_if(|(k, _)| *k == ka) {
1645                        self.ra.push(r);
1646                    }
1647                    while let Some((_, r)) = self.b.next_if(|(k, _)| *k == ka) {
1648                        self.rb.push(r);
1649                    }
1650                    self.i = 0;
1651                    self.j = 0;
1652                }
1653            }
1654        }
1655    }
1656}
1657
1658// --- index-nested-loop probing ------------------------------------------------
1659//
1660// Under a small, known demand (LIMIT/ASK — see `Ctx::limit_hint`), scanning
1661// every pattern once to hash-join is mostly wasted work: the consumer wants a
1662// handful of rows. The probe path instead streams the seed/first pattern and,
1663// per row, *probes* each next pattern through the index with the row's bound
1664// values substituted as scan constants — so producing k solutions touches
1665// O(k · patterns) index groups instead of every pattern's full extent. Same
1666// solution multiset as the hash join (joins are order-independent); only the
1667// evaluation order differs.
1668
1669/// Index-scan constant for a subject position given a partially-bound row.
1670/// Outer `None` = unsatisfiable for this row; inner `None` = wildcard.
1671fn probe_subject(ctx: &Ctx, t: &SlotTerm, base: &Row) -> Option<Option<u32>> {
1672    let dict = ctx.rete.dictionary();
1673    match t {
1674        SlotTerm::Node(n) => dict.node_as_subject_id(*n).map(Some),
1675        SlotTerm::Pred(_) => None,
1676        SlotTerm::Var(i) => match &base[*i] {
1677            None => Some(None),
1678            Some(Val::Id(v)) if *v >= 0 => dict.node_as_subject_id(*v as u32).map(Some),
1679            // A predicate-tagged or computed value can never be a subject.
1680            Some(_) => None,
1681        },
1682    }
1683}
1684
1685/// Index-scan constant for an object position (see [`probe_subject`]).
1686fn probe_object(ctx: &Ctx, t: &SlotTerm, base: &Row) -> Option<Option<u32>> {
1687    let dict = ctx.rete.dictionary();
1688    match t {
1689        SlotTerm::Node(n) => dict.node_as_object_id(*n).map(Some),
1690        SlotTerm::Pred(_) => None,
1691        SlotTerm::Var(i) => match &base[*i] {
1692            None => Some(None),
1693            Some(Val::Id(v)) if *v >= 0 => dict.node_as_object_id(*v as u32).map(Some),
1694            Some(_) => None,
1695        },
1696    }
1697}
1698
1699/// Index-scan constant for a predicate position (see [`probe_subject`]).
1700fn probe_predicate(ctx: &Ctx, t: &SlotTerm, base: &Row) -> Option<Option<u32>> {
1701    let dict = ctx.rete.dictionary();
1702    match t {
1703        SlotTerm::Pred(p) => Some(Some(*p)),
1704        SlotTerm::Node(_) => None,
1705        SlotTerm::Var(i) => match &base[*i] {
1706            None => Some(None),
1707            Some(Val::Id(v)) if *v < 0 => Some(Some((-v - 1) as u32)),
1708            // Canonicalized to a node id — the term may still be a predicate.
1709            Some(Val::Id(v)) => ctx
1710                .resolver
1711                .term(*v)
1712                .and_then(|t| dict.predicate_id(&t))
1713                .map(Some),
1714            Some(Val::Str(_)) => None,
1715        },
1716    }
1717}
1718
1719/// Probe one pattern with a partially-bound row: every bound variable becomes
1720/// an index-scan constant, and each matching triple extends a clone of the row
1721/// (repeated unbound variables stay consistent).
1722fn probe_rows<'q>(
1723    ctx: &'q Ctx<'q>,
1724    index: &'q GraphIndex,
1725    t: (SlotTerm, SlotTerm, SlotTerm),
1726    base: Row,
1727) -> Box<dyn Iterator<Item = Row> + 'q> {
1728    let (Some(sid), Some(pid), Some(oid)) = (
1729        probe_subject(ctx, &t.0, &base),
1730        probe_predicate(ctx, &t.1, &base),
1731        probe_object(ctx, &t.2, &base),
1732    ) else {
1733        return Box::new(std::iter::empty());
1734    };
1735    let dict = ctx.rete.dictionary();
1736    Box::new(
1737        index
1738            .scan_iter((sid, pid, oid))
1739            .filter_map(move |(s_id, p_id, o_id)| {
1740                let s_val = dict.subject_node(s_id) as i64;
1741                let p_val = pred_val(ctx, &t.1, p_id);
1742                let o_val = dict.object_node(o_id) as i64;
1743                let mut row = base.clone();
1744                for (term, val) in [(&t.0, s_val), (&t.1, p_val), (&t.2, o_val)] {
1745                    if let SlotTerm::Var(i) = term {
1746                        match row[*i] {
1747                            Some(Val::Id(existing)) if existing != val => return None,
1748                            Some(Val::Id(_)) => {}
1749                            Some(Val::Str(_)) => return None,
1750                            None => row[*i] = Some(Val::Id(val)),
1751                        }
1752                    }
1753                }
1754                Some(row)
1755            }),
1756    )
1757}
1758
1759/// A lowered, probe-ordered BGP, reusable across many seed rows.
1760pub(crate) struct ProbePlan {
1761    pats: Vec<(SlotTerm, SlotTerm, SlotTerm)>,
1762}
1763
1764impl ProbePlan {
1765    /// Lower and order `patterns` for probing from rows that bind (at least)
1766    /// the slots in `seed_mask`. `None` = a constant term is unknown, making
1767    /// the BGP unsatisfiable for every seed.
1768    pub(crate) fn new(
1769        ctx: &Ctx,
1770        index: &GraphIndex,
1771        patterns: &[TriplePattern],
1772        seed_mask: &[bool],
1773    ) -> Option<Self> {
1774        let lowered = lower(patterns, ctx)?;
1775        let seed: std::collections::HashSet<usize> = seed_mask
1776            .iter()
1777            .enumerate()
1778            .filter_map(|(i, b)| b.then_some(i))
1779            .collect();
1780        let order = selectivity_order_seeded(ctx, index, &lowered, &seed);
1781        Some(ProbePlan {
1782            pats: order.into_iter().map(|i| lowered[i]).collect(),
1783        })
1784    }
1785}
1786
1787/// Depth-first index-nested-loop join over a [`ProbePlan`], starting from a
1788/// seed row. Fully lazy: pulling k rows probes O(k · patterns) index groups.
1789pub(crate) struct ProbeJoin<'q> {
1790    ctx: &'q Ctx<'q>,
1791    index: &'q GraphIndex,
1792    pats: Vec<(SlotTerm, SlotTerm, SlotTerm)>,
1793    stack: Vec<Box<dyn Iterator<Item = Row> + 'q>>,
1794}
1795
1796impl<'q> ProbeJoin<'q> {
1797    /// Probe a whole BGP from scratch (the seed is the all-unbound row).
1798    /// `None` = unsatisfiable.
1799    pub(crate) fn new(
1800        ctx: &'q Ctx<'q>,
1801        index: &'q GraphIndex,
1802        patterns: &[TriplePattern],
1803    ) -> Option<Self> {
1804        let plan = ProbePlan::new(ctx, index, patterns, &vec![false; ctx.slots.len()])?;
1805        Some(Self::from_plan(ctx, index, &plan, ctx.slots.empty_row()))
1806    }
1807
1808    /// Probe a pre-lowered plan from one seed row.
1809    pub(crate) fn from_plan(
1810        ctx: &'q Ctx<'q>,
1811        index: &'q GraphIndex,
1812        plan: &ProbePlan,
1813        seed: Row,
1814    ) -> Self {
1815        let pats = plan.pats.clone();
1816        let first = probe_rows(ctx, index, pats[0], seed);
1817        ProbeJoin {
1818            ctx,
1819            index,
1820            pats,
1821            stack: vec![first],
1822        }
1823    }
1824}
1825
1826impl Iterator for ProbeJoin<'_> {
1827    type Item = Row;
1828
1829    fn next(&mut self) -> Option<Row> {
1830        loop {
1831            let depth = self.stack.len();
1832            match self.stack.last_mut()?.next() {
1833                Some(row) => {
1834                    if depth == self.pats.len() {
1835                        return Some(row);
1836                    }
1837                    let it = probe_rows(self.ctx, self.index, self.pats[depth], row);
1838                    self.stack.push(it);
1839                }
1840                None => {
1841                    self.stack.pop();
1842                    if self.stack.is_empty() {
1843                        return None;
1844                    }
1845                }
1846            }
1847        }
1848    }
1849}
1850
1851/// Hash-join two BGP relations on the slots they share. Every left row binds
1852/// exactly the slots accumulated so far and every right row binds the current
1853/// pattern's slots, so the shared set is uniform across rows.
1854fn hash_join(
1855    left: Vec<Row>,
1856    left_bound: &[usize],
1857    right: Vec<Row>,
1858    right_slots: &[usize],
1859) -> Vec<Row> {
1860    // Fast path: the all-unbound seed row joins to `right` unchanged.
1861    if left.len() == 1 && left_bound.is_empty() {
1862        return right;
1863    }
1864    if left.is_empty() || right.is_empty() {
1865        return Vec::new();
1866    }
1867    let shared: Vec<usize> = right_slots
1868        .iter()
1869        .copied()
1870        .filter(|s| left_bound.contains(s))
1871        .collect();
1872    let fill = |l: &Row, r: &Row| -> Row {
1873        let mut out = l.clone();
1874        for &s in right_slots {
1875            out[s] = r[s].clone();
1876        }
1877        out
1878    };
1879    if shared.is_empty() {
1880        // No shared variable: Cartesian product. In a connected BGP this only
1881        // arises for a fully-ground pattern (deduped index ⇒ one matching row),
1882        // so it acts as an existence filter rather than multiplying rows.
1883        let mut out = Vec::with_capacity(left.len() * right.len());
1884        for l in &left {
1885            for r in &right {
1886                out.push(fill(l, r));
1887            }
1888        }
1889        return out;
1890    }
1891    let key_of = |b: &Row| -> Vec<Val> { shared.iter().map(|&s| b[s].clone().unwrap()).collect() };
1892    // A hash join is symmetric on the shared key, so build the **smaller** side's
1893    // hash table and probe with the larger — same result set, fewer inserts and a
1894    // smaller table. (The old code always built `right`; after cardinality-based
1895    // ordering the accumulating `left` is usually the small selective side, so
1896    // building it instead is the common win.) Whichever side is built, the output
1897    // row carries every bound slot of both.
1898    let mut out = Vec::new();
1899    if right.len() <= left.len() {
1900        let mut buckets: HashMap<Vec<Val>, Vec<Row>> = HashMap::new();
1901        for r in right {
1902            buckets.entry(key_of(&r)).or_default().push(r);
1903        }
1904        for l in &left {
1905            if let Some(rs) = buckets.get(&key_of(l)) {
1906                for r in rs {
1907                    out.push(fill(l, r));
1908                }
1909            }
1910        }
1911    } else {
1912        // Build the left side; fill the left's own slots into each probed right.
1913        let mut buckets: HashMap<Vec<Val>, Vec<Row>> = HashMap::new();
1914        for l in left {
1915            buckets.entry(key_of(&l)).or_default().push(l);
1916        }
1917        for r in &right {
1918            if let Some(ls) = buckets.get(&key_of(r)) {
1919                for l in ls {
1920                    let mut row = r.clone();
1921                    for &s in left_bound {
1922                        if let Some(v) = &l[s] {
1923                            row[s] = Some(v.clone());
1924                        }
1925                    }
1926                    out.push(row);
1927                }
1928            }
1929        }
1930    }
1931    out
1932}
1933
1934// Constant-only index constraints. A variable scans as a wildcard (`Some(None)`)
1935// — it is resolved later by the hash join, not pushed into the scan. The outer
1936// `None` means "unsatisfiable" (a constant in an impossible role, or unknown to
1937// the dictionary), which empties the whole BGP. Inner `None` = "wildcard".
1938fn const_subject(t: &SlotTerm, d: &crate::Dictionary) -> Option<Option<u32>> {
1939    match t {
1940        SlotTerm::Node(n) => d.node_as_subject_id(*n).map(Some),
1941        SlotTerm::Pred(_) => None, // a predicate term can't be a subject
1942        SlotTerm::Var(_) => Some(None),
1943    }
1944}
1945
1946fn const_object(t: &SlotTerm, d: &crate::Dictionary) -> Option<Option<u32>> {
1947    match t {
1948        SlotTerm::Node(n) => d.node_as_object_id(*n).map(Some),
1949        SlotTerm::Pred(_) => None,
1950        SlotTerm::Var(_) => Some(None),
1951    }
1952}
1953
1954fn const_predicate(t: &SlotTerm) -> Option<Option<u32>> {
1955    match t {
1956        SlotTerm::Pred(p) => Some(Some(*p)),
1957        SlotTerm::Node(_) => None, // a node term can't be a predicate
1958        SlotTerm::Var(_) => Some(None),
1959    }
1960}
1961
1962#[cfg(test)]
1963mod tests {
1964    use super::*;
1965    use crate::dictionary::DictionaryBuilder;
1966    use crate::file::write_file_v05 as write_file;
1967    use crate::index::GraphIndexBuilder;
1968
1969    fn rete_from(triples: &[(&str, &str, &str)]) -> Vec<u8> {
1970        let mut db = DictionaryBuilder::new();
1971        for (s, p, o) in triples {
1972            db.observe(s, p, o);
1973        }
1974        let dict = db.build();
1975        let mut ib = GraphIndexBuilder::new();
1976        for (s, p, o) in triples {
1977            ib.push(dict.encode(s, p, o).unwrap());
1978        }
1979        write_file(&dict, &ib.build(), false, &[], 0)
1980    }
1981
1982    fn pat(s: &str, p: &str, o: &str) -> TriplePattern {
1983        TriplePattern {
1984            s: PatternTerm::parse(s),
1985            p: PatternTerm::parse(p),
1986            o: PatternTerm::parse(o),
1987        }
1988    }
1989
1990    /// The scan-bytes figure (`rete cost`'s preview, and through 0.3.7 the
1991    /// merge-join and fat-side gates) routes a bound term by its id IN ITS
1992    /// ROLE. It used to route the unified node id,
1993    /// which for an object-only term (any literal) is shifted past the whole
1994    /// subject-only block: `?s :value "lit"` found no OPS tile at all, cost
1995    /// "0 bytes", and so let a merge join materialize its partner in full —
1996    /// on OpenCitations, every DOI identifier of a 2.9 B-triple file, behind
1997    /// `?id :hasLiteralValue "10.…" ; :usesIdentifierScheme :doi`.
1998    #[test]
1999    fn scan_bytes_route_a_literal_by_its_object_id() {
2000        let value = "<http://ex/value>";
2001        let owned: Vec<(String, String)> = (0..400)
2002            .map(|i| (format!("<http://ex/s{i:03}>"), format!("\"lit {i:03}\"")))
2003            .collect();
2004        let mut db = DictionaryBuilder::new();
2005        for (s, o) in &owned {
2006            db.observe(s, value, o);
2007        }
2008        let dict = db.build();
2009        let mut ib = GraphIndexBuilder::new().with_tile_budget(256);
2010        for (s, o) in &owned {
2011            ib.push(dict.encode(s, value, o).unwrap());
2012        }
2013        let rete = Rete::open(&write_file(&dict, &ib.build(), false, &[], 0)).unwrap();
2014        let index = rete.default_index();
2015        // The whole predicate, the most a bound object could ever cost.
2016        let whole = routed_scan_bytes(index, (None, rete.dictionary().predicate_id(value), None));
2017        for (_, lit) in owned.iter().step_by(57) {
2018            let q = [pat("?s", value, lit)];
2019            let mut slots = Slots::new();
2020            collect_pattern_slots(&q, &mut slots);
2021            let ctx = Ctx::new(&rete, slots);
2022            let got = bgp_min_scan_bytes(&ctx, index, &q).unwrap();
2023            assert!(got > 0, "{lit} routed to no tile");
2024            assert!(
2025                got < whole,
2026                "{lit} routed to the whole predicate ({got} of {whole})"
2027            );
2028        }
2029    }
2030
2031    /// The planner's size gates read ROW estimates, which do not depend on how
2032    /// the tiles are compressed. The same file opened resident (tile sizes are
2033    /// the decoded images) and lazily (tile sizes are the compressed frames)
2034    /// reports different bytes but must take the same decisions — the byte
2035    /// gates this replaced flipped plans when only the zstd level changed.
2036    #[test]
2037    fn plan_gates_do_not_move_with_the_compression() {
2038        let mut owned: Vec<(String, String, String)> = Vec::new();
2039        for i in 0..3000 {
2040            let s = format!("<http://ex/s{i}>");
2041            owned.push((
2042                s.clone(),
2043                "<http://ex/kind>".into(),
2044                format!("<http://ex/K{}>", i % 7),
2045            ));
2046            owned.push((s.clone(), "<http://ex/value>".into(), format!("\"v {i}\"")));
2047            if i % 50 == 0 {
2048                owned.push((s, "<http://ex/rare>".into(), "<http://ex/R>".into()));
2049            }
2050        }
2051        let mut db = DictionaryBuilder::new();
2052        for (s, p, o) in &owned {
2053            db.observe(s, p, o);
2054        }
2055        let dict = db.build();
2056        let mut ib = GraphIndexBuilder::new().with_tile_budget(1024);
2057        for (s, p, o) in &owned {
2058            ib.push(dict.encode(s, p, o).unwrap());
2059        }
2060        let bytes = write_file(&dict, &ib.build(), false, &[], 0);
2061        let leaked: &'static [u8] = Box::leak(bytes.clone().into_boxed_slice());
2062        let resident = Rete::open(&bytes).unwrap();
2063        let lazy = Rete::open_ranged_lazy(crate::reader::SliceReader::new(leaked)).unwrap();
2064        let shapes: Vec<Vec<TriplePattern>> = vec![
2065            vec![
2066                pat("?s", "<http://ex/rare>", "<http://ex/R>"),
2067                pat("?s", "<http://ex/value>", "?v"),
2068            ],
2069            vec![
2070                pat("?s", "<http://ex/kind>", "<http://ex/K3>"),
2071                pat("?s", "<http://ex/value>", "?v"),
2072            ],
2073            vec![
2074                pat("?s", "<http://ex/kind>", "?k"),
2075                pat("?s", "<http://ex/value>", "?v"),
2076            ],
2077            vec![
2078                pat("?s", "<http://ex/value>", "\"v 7\""),
2079                pat("?s", "<http://ex/kind>", "?k"),
2080            ],
2081        ];
2082        let decisions = |rete: &Rete| -> Vec<(Vec<u64>, bool, Option<u64>)> {
2083            shapes
2084                .iter()
2085                .map(|q| {
2086                    let mut slots = Slots::new();
2087                    collect_pattern_slots(q, &mut slots);
2088                    let ctx = Ctx::new(rete, slots);
2089                    let index = rete.default_index();
2090                    let lowered = lower(q, &ctx).unwrap();
2091                    let est = tile_estimates(&ctx, index, &lowered, &Default::default());
2092                    let lopsided = merge_seed_lopsided(&ctx, index, &lowered[0], &lowered[1]);
2093                    let min = bgp_min_estimate(&ctx, index, q).map(|e| e as u64);
2094                    (est.iter().map(|e| *e as u64).collect(), lopsided, min)
2095                })
2096                .collect()
2097        };
2098        assert_eq!(decisions(&resident), decisions(&lazy));
2099        // ... while the byte figures the gates used to read do differ.
2100        #[cfg(feature = "compression")]
2101        {
2102            let bytes_of = |rete: &Rete| {
2103                let q = &shapes[2];
2104                let mut slots = Slots::new();
2105                collect_pattern_slots(q, &mut slots);
2106                let ctx = Ctx::new(rete, slots);
2107                bgp_min_scan_bytes(&ctx, rete.default_index(), q)
2108            };
2109            assert_ne!(bytes_of(&resident), bytes_of(&lazy));
2110        }
2111    }
2112
2113    /// A fat predicate must look fat whichever layout stores the directory.
2114    /// On format 0x06 the tile directory is two-level and a leaf that is not
2115    /// resident is estimated from the root alone; it used to count as ONE
2116    /// tile, so a 2-million-row `rdfs:label` read as ~7 thousand rows on 0x06
2117    /// and 2 million on 0x05 — and the OPTIONAL joined to a 1000-row left side
2118    /// was hash-joined (the whole predicate materialized: 14.5 GB on a 8 B
2119    /// triple file) instead of probed per row. The same file in both layouts
2120    /// must give estimates of the same order, and the probe decision must be
2121    /// the same one.
2122    #[test]
2123    fn a_fat_predicate_estimates_alike_in_both_layouts() {
2124        let mut owned: Vec<(String, String, String)> = Vec::new();
2125        for i in 0..120_000 {
2126            let s = format!("<http://ex/s{i:06}>");
2127            owned.push((s.clone(), "<http://ex/kind>".into(), "<http://ex/K>".into()));
2128            owned.push((s, "<http://ex/value>".into(), format!("\"v {i}\"")));
2129        }
2130        let mut db = DictionaryBuilder::new();
2131        for (s, p, o) in &owned {
2132            db.observe(s, p, o);
2133        }
2134        let dict = db.build();
2135        let mut ib = GraphIndexBuilder::new().with_tile_budget(256);
2136        for (s, p, o) in &owned {
2137            ib.push(dict.encode(s, p, o).unwrap());
2138        }
2139        let v5 = write_file(&dict, &ib.build(), false, &[], 0);
2140        let v6: &'static [u8] = Box::leak(
2141            crate::v06::write::convert_image(&v5, Default::default())
2142                .unwrap()
2143                .into_boxed_slice(),
2144        );
2145        let leaked5: &'static [u8] = Box::leak(v5.clone().into_boxed_slice());
2146        let flat = Rete::open_ranged_lazy(crate::reader::SliceReader::new(leaked5)).unwrap();
2147        let leafed = Rete::open_ranged_lazy(crate::reader::SliceReader::new(v6)).unwrap();
2148        let q = [pat("?s", "<http://ex/value>", "?v")];
2149        let figures = |rete: &Rete| {
2150            let mut slots = Slots::new();
2151            collect_pattern_slots(&q, &mut slots);
2152            let ctx = Ctx::new(rete, slots);
2153            let index = rete.default_index();
2154            let total = bgp_total_estimate(&ctx, index, &q).unwrap();
2155            (total, probe_row_budget(total))
2156        };
2157        let (est5, budget5) = figures(&flat);
2158        let (est6, budget6) = figures(&leafed);
2159        assert!(est5 >= 70_000.0, "flat estimate {est5}");
2160        assert!(
2161            est6 >= est5 / 2.0 && est6 <= est5 * 2.0,
2162            "0x06 estimates {est6} where 0x05 estimates {est5}"
2163        );
2164        // The decision: a 1000-row left side probes in both.
2165        assert!(budget5.is_some_and(|b| b >= 1000), "{budget5:?}");
2166        assert!(budget6.is_some_and(|b| b >= 1000), "{budget6:?}");
2167    }
2168
2169    /// A selective object on a fat leading predicate must estimate small in
2170    /// both layouts: `?s rdf:type <rare class>` is a few tiles of a predicate
2171    /// that fills thousands, and the planner seeds the join from it. On 0x06 a
2172    /// leaf that is not resident used to be estimated without the bound object
2173    /// (the whole of `rdf:type`) or by the ids' average share of its leaf
2174    /// (skewed ids put a rare class beside a huge one).
2175    #[test]
2176    fn a_rare_object_of_a_fat_predicate_estimates_small_in_both_layouts() {
2177        let ty = "<http://www.w3.org/1999/02/22-rdf-syntax-ns#type>";
2178        let mut owned: Vec<(String, String, String)> = Vec::new();
2179        for i in 0..120_000 {
2180            let s = format!("<http://ex/s{i:06}>");
2181            owned.push((s.clone(), ty.into(), "<http://ex/Common>".into()));
2182            owned.push((s.clone(), "<http://ex/other>".into(), format!("\"o {i}\"")));
2183            if i % 3000 == 0 {
2184                owned.push((s, ty.into(), "<http://ex/Rare>".into()));
2185            }
2186        }
2187        let mut db = DictionaryBuilder::new();
2188        for (s, p, o) in &owned {
2189            db.observe(s, p, o);
2190        }
2191        let dict = db.build();
2192        let mut ib = GraphIndexBuilder::new().with_tile_budget(256);
2193        for (s, p, o) in &owned {
2194            ib.push(dict.encode(s, p, o).unwrap());
2195        }
2196        let v5 = write_file(&dict, &ib.build(), false, &[], 0);
2197        let v6: &'static [u8] = Box::leak(
2198            crate::v06::write::convert_image(&v5, Default::default())
2199                .unwrap()
2200                .into_boxed_slice(),
2201        );
2202        let leaked5: &'static [u8] = Box::leak(v5.into_boxed_slice());
2203        let figures = |rete: &Rete| {
2204            let est = |q: &[TriplePattern]| {
2205                let mut slots = Slots::new();
2206                collect_pattern_slots(q, &mut slots);
2207                let ctx = Ctx::new(rete, slots);
2208                bgp_min_estimate(&ctx, rete.default_index(), q).unwrap()
2209            };
2210            (
2211                est(&[pat("?s", ty, "<http://ex/Rare>")]),
2212                est(&[pat("?s", ty, "<http://ex/Common>")]),
2213                est(&[pat("?s", "<http://ex/other>", "?x")]),
2214            )
2215        };
2216        for (name, bytes) in [("0x05", leaked5), ("0x06", v6)] {
2217            let rete = Rete::open_ranged_lazy(crate::reader::SliceReader::new(bytes)).unwrap();
2218            let (rare, common, other) = figures(&rete);
2219            assert!(rare * 20.0 < other, "{name}: rare {rare} vs other {other}");
2220            assert!(
2221                rare * 20.0 < common,
2222                "{name}: rare {rare} vs common {common}"
2223            );
2224        }
2225    }
2226
2227    /// The run of tiles holding a leading id is the same in both layouts, for
2228    /// every id and every permutation, and a leafed lookup does not walk the
2229    /// leaf tile by tile (a probe join runs it per row: 94,700 probes took 126 s
2230    /// on a 0x06 file against 1.3 s on its 0x05 copy).
2231    #[test]
2232    fn a_tile_span_is_the_same_in_both_layouts() {
2233        let ty = "<http://www.w3.org/1999/02/22-rdf-syntax-ns#type>";
2234        let mut owned: Vec<(String, String, String)> = Vec::new();
2235        for i in 0..30_000 {
2236            let s = format!("<http://ex/s{i:06}>");
2237            owned.push((s.clone(), ty.into(), format!("<http://ex/C{}>", i % 7)));
2238            owned.push((s, "<http://ex/other>".into(), format!("\"o {}\"", i % 5000)));
2239        }
2240        let mut db = DictionaryBuilder::new();
2241        for (s, p, o) in &owned {
2242            db.observe(s, p, o);
2243        }
2244        let dict = db.build();
2245        let mut ib = GraphIndexBuilder::new().with_tile_budget(256);
2246        for (s, p, o) in &owned {
2247            ib.push(dict.encode(s, p, o).unwrap());
2248        }
2249        let v5 = write_file(&dict, &ib.build(), false, &[], 0);
2250        let v6 = crate::v06::write::convert_image(&v5, Default::default()).unwrap();
2251        let (v5, v6): (&'static [u8], &'static [u8]) = (
2252            Box::leak(v5.into_boxed_slice()),
2253            Box::leak(v6.into_boxed_slice()),
2254        );
2255        let flat = Rete::open_ranged_lazy(crate::reader::SliceReader::new(v5)).unwrap();
2256        let leafed = Rete::open_ranged_lazy(crate::reader::SliceReader::new(v6)).unwrap();
2257        let (fi, li) = (flat.default_index(), leafed.default_index());
2258        let mut checked = 0;
2259        for perm in fi.perms().iter() {
2260            let si = perm.section_index();
2261            for a in (0..dict.term_count() + 2).step_by(97) {
2262                assert_eq!(
2263                    fi.tile_span(si, Some(a)),
2264                    li.tile_span(si, Some(a)),
2265                    "section {si}, id {a}"
2266                );
2267                checked += 1;
2268            }
2269            assert_eq!(fi.tile_span(si, None), li.tile_span(si, None));
2270        }
2271        assert!(checked > 100);
2272    }
2273
2274    /// Every join path the planner can pick — materialized prefix, probed last
2275    /// pattern, streamed seed with hashed or probed later patterns, streaming
2276    /// merge join — returns the same solution multiset. The thresholds are
2277    /// shrunk so a 400-triple graph takes each path.
2278    #[test]
2279    fn every_join_path_returns_the_same_solutions() {
2280        let mut owned: Vec<(String, String, String)> = Vec::new();
2281        let mut x: u64 = 7;
2282        let mut next = |m: u64| {
2283            x = x
2284                .wrapping_mul(6364136223846793005)
2285                .wrapping_add(1442695040888963407);
2286            (x >> 33) % m
2287        };
2288        for i in 0..60 {
2289            let s = format!("s{i}");
2290            owned.push((s.clone(), "a".into(), format!("C{}", next(3))));
2291            for p in 0..4 {
2292                for _ in 0..next(3) {
2293                    owned.push((s.clone(), format!("p{p}"), format!("s{}", next(60))));
2294                }
2295            }
2296        }
2297        let triples: Vec<(&str, &str, &str)> = owned
2298            .iter()
2299            .map(|(s, p, o)| (s.as_str(), p.as_str(), o.as_str()))
2300            .collect();
2301        let bytes = rete_from(&triples);
2302        let rete = Rete::open(&bytes).unwrap();
2303        let queries: Vec<Vec<TriplePattern>> = vec![
2304            vec![pat("?s", "p0", "?a"), pat("?s", "p1", "?b")],
2305            vec![pat("?s", "a", "C1"), pat("?s", "p0", "?a")],
2306            vec![pat("?s", "p0", "?a"), pat("?a", "p2", "?b")],
2307            vec![pat("s5", "p0", "?a"), pat("?a", "?p", "?o")],
2308            vec![pat("?s", "p0", "?x"), pat("?s", "p1", "?x")],
2309            vec![
2310                pat("?s", "p3", "?o"),
2311                pat("?o", "a", "C0"),
2312                pat("?s", "p1", "?z"),
2313            ],
2314            vec![
2315                pat("?s", "p0", "?a"),
2316                pat("?s", "p1", "?b"),
2317                pat("?b", "p2", "?c"),
2318            ],
2319        ];
2320        let run = |limits: Option<PlanLimits>| -> Vec<Vec<Vec<(String, String)>>> {
2321            TEST_LIMITS.with(|c| c.set(limits));
2322            let out = queries
2323                .iter()
2324                .map(|q| {
2325                    let mut slots = crate::row::Slots::new();
2326                    collect_pattern_slots(q, &mut slots);
2327                    let ctx = Ctx::new(&rete, slots);
2328                    let mut rows: Vec<Vec<(String, String)>> =
2329                        BgpSolutions::new(&ctx, rete.default_index(), q)
2330                            .map(|r| row_to_binding(&ctx, &r).into_iter().collect())
2331                            .collect();
2332                    rows.sort();
2333                    rows
2334                })
2335                .collect();
2336            TEST_LIMITS.with(|c| c.set(None));
2337            out
2338        };
2339        let reference = run(None);
2340        assert!(reference.iter().all(|r| !r.is_empty()), "{reference:?}");
2341        for (name, l) in [
2342            (
2343                "probe everything",
2344                PlanLimits {
2345                    probe_floor: 0.0,
2346                    stream_seed: f64::MAX,
2347                    hash_build: f64::MAX,
2348                },
2349            ),
2350            (
2351                "stream + hash",
2352                PlanLimits {
2353                    probe_floor: f64::MAX,
2354                    stream_seed: 0.0,
2355                    hash_build: f64::MAX,
2356                },
2357            ),
2358            (
2359                "stream + probe",
2360                PlanLimits {
2361                    probe_floor: f64::MAX,
2362                    stream_seed: 0.0,
2363                    hash_build: 0.0,
2364                },
2365            ),
2366            (
2367                "stream, all paths",
2368                PlanLimits {
2369                    probe_floor: 0.0,
2370                    stream_seed: 0.0,
2371                    hash_build: 30.0,
2372                },
2373            ),
2374        ] {
2375            assert_eq!(run(Some(l)), reference, "{name}");
2376        }
2377    }
2378
2379    #[test]
2380    fn single_pattern_binds_variable() {
2381        let bytes = rete_from(&[("Alice", "knows", "Bob"), ("Bob", "knows", "Carol")]);
2382        let rete = Rete::open(&bytes).unwrap();
2383        let sols = eval_bgp(&rete, &[pat("Alice", "knows", "?y")]);
2384        assert_eq!(sols.len(), 1);
2385        assert_eq!(sols[0]["y"], "Bob");
2386    }
2387
2388    #[test]
2389    fn two_hop_join_on_shared_variable() {
2390        // Alice -> Bob -> Carol, plus a dead-end branch.
2391        let bytes = rete_from(&[
2392            ("Alice", "knows", "Bob"),
2393            ("Bob", "knows", "Carol"),
2394            ("Carol", "knows", "Dave"),
2395            ("Alice", "knows", "Eve"), // Eve knows no one -> no 2-hop
2396        ]);
2397        let rete = Rete::open(&bytes).unwrap();
2398        let sols = eval_bgp(&rete, &[pat("?x", "knows", "?y"), pat("?y", "knows", "?z")]);
2399        // Alice-Bob-Carol, Bob-Carol-Dave.
2400        let mut got: Vec<_> = sols
2401            .iter()
2402            .map(|b| (b["x"].clone(), b["y"].clone(), b["z"].clone()))
2403            .collect();
2404        got.sort();
2405        assert_eq!(
2406            got,
2407            vec![
2408                ("Alice".into(), "Bob".into(), "Carol".into()),
2409                ("Bob".into(), "Carol".into(), "Dave".into()),
2410            ]
2411        );
2412    }
2413
2414    #[test]
2415    fn repeated_variable_within_pattern() {
2416        // A mutual/self relation: only Bob knows himself.
2417        let bytes = rete_from(&[("Alice", "knows", "Bob"), ("Bob", "knows", "Bob")]);
2418        let rete = Rete::open(&bytes).unwrap();
2419        let sols = eval_bgp(&rete, &[pat("?x", "knows", "?x")]);
2420        assert_eq!(sols.len(), 1);
2421        assert_eq!(sols[0]["x"], "Bob");
2422    }
2423
2424    #[test]
2425    fn no_solutions_yields_empty() {
2426        let bytes = rete_from(&[("Alice", "knows", "Bob")]);
2427        let rete = Rete::open(&bytes).unwrap();
2428        assert!(eval_bgp(&rete, &[pat("Alice", "likes", "?y")]).is_empty());
2429    }
2430
2431    /// A LUBM-Q7-shaped snowflake: a popular class enumeration (`?x a Student`)
2432    /// plus a selective adjacency seed (`<Prof> teacherOf ?y`). Exercises both the
2433    /// **class-enum seed deprioritization** (the join must not start from the huge
2434    /// `a Student` extent) and the **probe-the-tail hybrid** (the type checks are
2435    /// applied late, per row). The answer must equal a brute-force computation
2436    /// regardless of how the join is ordered/probed.
2437    #[test]
2438    fn snowflake_type_and_adjacency_matches_reference() {
2439        const TYPE: &str = "<http://www.w3.org/1999/02/22-rdf-syntax-ns#type>";
2440        let mut triples: Vec<(String, String, String)> = Vec::new();
2441        // 60 students, each a Student and taking 2 courses (c{i%5}, c{(i+1)%5}).
2442        for i in 0..60 {
2443            let s = format!("S{i}");
2444            triples.push((s.clone(), TYPE.into(), "Student".into()));
2445            triples.push((s.clone(), "takesCourse".into(), format!("c{}", i % 5)));
2446            triples.push((s, "takesCourse".into(), format!("c{}", (i + 1) % 5)));
2447        }
2448        // 5 courses; the professor teaches only c0 and c2.
2449        for j in 0..5 {
2450            triples.push((format!("c{j}"), TYPE.into(), "Course".into()));
2451        }
2452        triples.push(("Prof".into(), "teacherOf".into(), "c0".into()));
2453        triples.push(("Prof".into(), "teacherOf".into(), "c2".into()));
2454
2455        let refs: Vec<(&str, &str, &str)> = triples
2456            .iter()
2457            .map(|(s, p, o)| (s.as_str(), p.as_str(), o.as_str()))
2458            .collect();
2459        let bytes = rete_from(&refs);
2460        let rete = Rete::open(&bytes).unwrap();
2461
2462        let got: std::collections::BTreeSet<(String, String)> = eval_bgp(
2463            &rete,
2464            &[
2465                pat("?x", TYPE, "Student"),
2466                pat("?y", TYPE, "Course"),
2467                pat("?x", "takesCourse", "?y"),
2468                pat("Prof", "teacherOf", "?y"),
2469            ],
2470        )
2471        .into_iter()
2472        .map(|b| (b["x"].clone(), b["y"].clone()))
2473        .collect();
2474
2475        // Brute force: students taking a course the professor teaches (c0 or c2).
2476        let taught = ["c0", "c2"];
2477        let mut want = std::collections::BTreeSet::new();
2478        for i in 0..60 {
2479            for c in [i % 5, (i + 1) % 5] {
2480                let course = format!("c{c}");
2481                if taught.contains(&course.as_str()) {
2482                    want.insert((format!("S{i}"), course));
2483                }
2484            }
2485        }
2486        assert_eq!(got, want, "snowflake join must match the brute-force set");
2487        assert!(!want.is_empty(), "sanity: the reference set is non-empty");
2488    }
2489
2490    /// A subject-subject join (`?x a ?av . ?x b ?bv`) is the same-role shape the
2491    /// merge join handles: both patterns sort on the subject column. With multiple
2492    /// `a` and `b` values per subject it must emit the full cross product per
2493    /// subject (equal-key run handling) and drop subjects missing either side.
2494    #[test]
2495    fn merge_join_subject_star_cross_product() {
2496        let bytes = rete_from(&[
2497            ("x", "a", "a1"),
2498            ("x", "a", "a2"),
2499            ("x", "b", "b1"),
2500            ("x", "b", "b2"),
2501            ("y", "a", "a3"),
2502            ("y", "b", "b3"),
2503            ("z", "a", "a4"), // z has no `b` → contributes no join row
2504        ]);
2505        let rete = Rete::open(&bytes).unwrap();
2506        let mut got: Vec<(String, String, String)> =
2507            eval_bgp(&rete, &[pat("?x", "a", "?av"), pat("?x", "b", "?bv")])
2508                .iter()
2509                .map(|m| (m["x"].clone(), m["av"].clone(), m["bv"].clone()))
2510                .collect();
2511        got.sort();
2512        let mut want: Vec<(String, String, String)> = [
2513            ("x", "a1", "b1"),
2514            ("x", "a1", "b2"),
2515            ("x", "a2", "b1"),
2516            ("x", "a2", "b2"),
2517            ("y", "a3", "b3"),
2518        ]
2519        .into_iter()
2520        .map(|(a, b, c)| (a.to_string(), b.to_string(), c.to_string()))
2521        .collect();
2522        want.sort();
2523        assert_eq!(got, want, "same-role merge must equal the brute-force join");
2524    }
2525}