rete_core/

sparql.rs

//! SPARQL parsing, lowering, and evaluation (SPEC.md §8).
//!
//! Queries are parsed with `spargebra`, lowered into a small plan algebra, and
//! evaluated against the `.rete` indexes. BGPs use the integer-space hash-join
//! engine in [`bgp`]; filters, joins, OPTIONAL, UNION, MINUS, VALUES, property
//! paths, named graphs, aggregates, and query forms are handled in the sibling
//! lowering/evaluation modules. Unsupported features are rejected explicitly
//! rather than silently dropped.
//!
//! [`bgp`]: crate::bgp

use spargebra::Query;

use crate::bgp::{Binding, PatternTerm, TriplePattern};
use crate::file::Rete;

mod aggregate;
mod diag;
mod distinct;
mod distinct_union;
mod eval;
mod expr;
mod lower;
mod partition;
mod path;
mod prefetch;
mod ql;
mod sparql12;
mod value;

pub use diag::{warnings_json, QueryWarning, WarningSeverity};
use eval::{ask_solution, instantiate, raw_solutions, run_select, run_select_communities};
use lower::{lower_pattern, lower_select, parse_query};
pub use lower::{parse_select, query_predicates};
pub use partition::PartitionPlan;
// Re-exported so the sibling modules' `use super::*` can reach the evaluator
// (expr's FILTER EXISTS evaluates a sub-plan via `eval_plan_in`).
pub(crate) use eval::eval_plan_in;

#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum SparqlError {
    #[error("parse error: {0}")]
    Parse(String),
    #[error("unsupported query feature: {0}")]
    Unsupported(&'static str),
    /// A function call whose IRI is neither a SPARQL built-in nor one of the
    /// extension functions rete implements (XSD casts, GeoSPARQL `geof:`, `geo3:`).
    #[error("unsupported query feature: function <{0}> is not implemented")]
    UnsupportedFunction(String),
    /// A (non-SILENT) `SERVICE` block failed: the endpoint errored, returned
    /// unparseable results, or no [`ServiceClient`](crate::ServiceClient) is
    /// attached to the file handle. Partial results are never returned.
    #[error("SERVICE federation: {0}")]
    Service(String),
    /// [`partition_plan`] / [`eval_query_partitioned`] refused to split a
    /// query by a variable, because the chunks' answers together would not be
    /// the query's answer; the message says why and what to do instead.
    #[error("cannot split this query {0}")]
    Partition(String),
}

/// A lowered SELECT query: solution modifiers plus the evaluation plan tree.
#[derive(Debug, Clone)]
pub struct Select {
    /// Projected variable names (empty = SELECT *).
    pub project: Vec<String>,
    /// DISTINCT (or REDUCED) requested.
    pub distinct: bool,
    /// OFFSET (solutions to skip).
    pub offset: usize,
    /// LIMIT (max solutions), if any.
    pub limit: Option<usize>,
    /// GROUP BY + aggregates, applied after the plan and before projection.
    pub group: Option<GroupSpec>,
    /// BIND/aggregate-alias assignments `(result_var, expr)`, applied after
    /// aggregation (e.g. `(COUNT(?f) AS ?n)` aliases an internal var to `?n`).
    pub extends: Vec<(String, FExpr)>,
    /// ORDER BY keys `(expr, descending)`, applied before projection/slice.
    pub order: Vec<(FExpr, bool)>,
    /// HAVING conditions, applied after aggregation.
    pub having: Vec<FExpr>,
    /// `FROM <iri>` graphs: when non-empty, the query's default graph is the
    /// union (RDF merge) of these named graphs.
    pub from: Vec<String>,
    /// `FROM NAMED <iri>` graphs: when `Some`, `GRAPH` may only see these.
    pub from_named: Option<Vec<String>>,
    /// **Union default graph** (opt-in, non-standard): when set and the query
    /// has no `FROM` of its own, the active default graph becomes the RDF merge
    /// of the file's default graph and *every* named graph — the mode Virtuoso,
    /// GraphDB and Jena TDB (`tdb:unionDefaultGraph`) offer. Plain SPARQL says a
    /// pattern outside `GRAPH` matches only the real default graph, so this is
    /// NEVER set by the parser — only by [`eval_query_with`] when the caller
    /// asked for it. An explicit `FROM` wins (the query named its own dataset).
    pub union_default: bool,
    /// The graph-pattern evaluation plan.
    pub plan: Plan,
    /// Monotonic counter for minting query-unique fresh variable names while
    /// lowering (e.g. the `?__qtN` holders that desugar RDF-star quoted-triple
    /// patterns). Kept on `Select` so it stays unique across every BGP in one
    /// query — two joined BGPs must not both mint `__qt1` and get unified.
    pub star_counter: usize,
}

impl Default for Select {
    fn default() -> Self {
        Select {
            project: Vec::new(),
            distinct: false,
            offset: 0,
            limit: None,
            group: None,
            extends: Vec::new(),
            order: Vec::new(),
            having: Vec::new(),
            from: Vec::new(),
            from_named: None,
            union_default: false,
            plan: Plan::Bgp(Vec::new()),
            star_counter: 0,
        }
    }
}

/// GROUP BY specification: grouping variables and result aggregates.
#[derive(Debug, Clone)]
pub struct GroupSpec {
    /// Variables to group by (empty = single group over all solutions).
    pub by: Vec<String>,
    /// `(result_variable, aggregate)` pairs.
    pub aggs: Vec<(String, Agg)>,
    /// Synthetic per-row columns computed *before* grouping — `(var, expr)` for
    /// each aggregate whose argument is an expression rather than a bare variable
    /// (e.g. `SUM(?a * 2)` lowers to a hidden `__aggN = ?a * 2`, then `SUM(__aggN)`).
    /// Empty for the common case (aggregating a plain variable).
    pub pre: Vec<(String, FExpr)>,
}

/// A supported aggregate function.
#[derive(Debug, Clone)]
pub enum Agg {
    /// COUNT(*) — number of solutions in the group.
    CountStar {
        distinct: bool,
    },
    /// COUNT(?v) — number of (optionally distinct) bound values.
    Count(String, bool),
    Sum(String),
    Avg(String),
    Min(String),
    Max(String),
    /// SAMPLE(?v) — any one value from the group.
    Sample(String),
    /// GROUP_CONCAT(\[DISTINCT\] ?v; SEPARATOR=...) — values joined by the
    /// separator (deduplicated when `distinct`).
    GroupConcat(String, String, bool),
}

/// A SPARQL graph-pattern evaluation plan (the supported algebra subset).
#[derive(Debug, Clone)]
#[must_use]
pub enum Plan {
    /// Basic graph pattern: triple patterns joined on shared variables.
    Bgp(Vec<TriplePattern>),
    /// Conjunction of two patterns (inner join on shared variables).
    Join(Box<Plan>, Box<Plan>),
    /// UNION: all solutions of either side.
    Union(Box<Plan>, Box<Plan>),
    /// OPTIONAL (left join): left solutions, extended by the right where it
    /// matches (and passes the optional condition), kept as-is where it doesn't.
    LeftJoin(Box<Plan>, Box<Plan>, Option<FExpr>),
    /// FILTER over an inner pattern.
    Filter(FExpr, Box<Plan>),
    /// In-pattern `BIND(expr AS ?var)`: each inner solution extended with `?var`
    /// (left unbound where `expr` errors). Distinct from the projection-time
    /// alias list (`Select::extends`) — this one is *inside* the graph pattern,
    /// so a following FILTER or join sees the bound variable.
    Extend(String, FExpr, Box<Plan>),
    /// A property path `subject <path> object`.
    Path(PatternTerm, PathAst, PatternTerm),
    /// Inline `VALUES`: variable names and rows of optional ground-term tokens
    /// (`None` = UNDEF).
    Values(Vec<String>, Vec<Vec<Option<String>>>),
    /// `MINUS`: left solutions, minus those compatible with a right solution
    /// that shares at least one bound variable.
    Minus(Box<Plan>, Box<Plan>),
    /// `GRAPH <iri>|?g { … }` — evaluate the inner pattern against a named graph.
    Graph(GraphTarget, Box<Plan>),
    /// The **distinct** bindings of `vars` over the union of single triple
    /// patterns (`sources`) — each solution once, however many sources (or
    /// triples of one source) produce it. Every variable of a source that is
    /// not in `vars` is anonymous: matched, never bound in the output. With
    /// `vars` empty it is an existence test (one empty solution or none).
    ///
    /// Built only by the OWL 2 QL rewriter (`ql`): an entailed atom such as
    /// `?x a C` holds once per entailed triple, while its rewriting
    /// (`?x a C₁`, `?x a C₂`, `?x p ?_` for a `p` with domain `C`, …) matches
    /// it once per *asserted* triple. The sources stream sorted on the key
    /// and are merged, so deduplicating costs one cursor per source, not a
    /// set of every solution.
    DistinctUnion {
        vars: Vec<String>,
        sources: Vec<TriplePattern>,
    },
    /// A nested `SELECT` subquery: evaluated independently to its projected
    /// solutions, which then join with the surrounding pattern on shared
    /// variables (only the subquery's projected variables are visible outside).
    Subquery(Box<Select>),
    /// `SERVICE [SILENT] <endpoint> { … }` — SPARQL 1.1 federated query. The
    /// inner pattern is shipped (as `query`, its re-serialized SPARQL text) to
    /// the remote endpoint through the file's attached
    /// [`ServiceClient`](crate::ServiceClient) at evaluation time; the returned
    /// solutions join the surrounding pattern on shared variables like any
    /// other operand. Under `silent`, a failed call degrades to one empty
    /// solution (per the spec); otherwise it fails the whole query.
    Service {
        silent: bool,
        /// The endpoint IRI (no angle brackets) — where the sub-query is sent.
        endpoint: String,
        /// Variables the inner pattern can bind (an over-approximation is fine:
        /// unreturned variables just stay unbound).
        vars: Vec<String>,
        /// The inner pattern as a standalone `SELECT`, exactly what is sent.
        query: String,
    },
}

/// The target of a `GRAPH` block.
#[derive(Debug, Clone)]
pub enum GraphTarget {
    /// `GRAPH <iri>` — one specific named graph.
    Named(String),
    /// `GRAPH ?g` — every named graph, binding the variable to its IRI.
    Var(String),
}

/// Path repetition operator.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Rep {
    /// exactly one hop (a plain predicate)
    One,
    /// `+` — one or more hops (transitive closure)
    OneOrMore,
    /// `*` — zero or more hops (reflexive-transitive closure)
    ZeroOrMore,
    /// `?` — zero or one hop
    ZeroOrOne,
}

/// A lowered property path expression, evaluated as a binary relation over the
/// graph's nodes.
#[derive(Debug, Clone)]
pub enum PathAst {
    /// A single predicate, optionally reversed.
    Pred(String, bool),
    /// Repetition (`*`/`+`/`?`) of a sub-path.
    Rep(Box<PathAst>, Rep),
    /// Sequence `a/b` (relational composition).
    Seq(Box<PathAst>, Box<PathAst>),
    /// Alternative `a|b` (union).
    Alt(Box<PathAst>, Box<PathAst>),
    /// Negated property set `!(p1|…|pn)` — one step over any predicate **not**
    /// in the set, in the given direction (`reversed` for the `^p` members,
    /// which `spargebra` wraps in a `Reverse`).
    NegatedSet(Vec<String>, bool),
}

/// Comparison operators supported in FILTER.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Op {
    Eq,
    Ne,
    Lt,
    Le,
    Gt,
    Ge,
}

/// Arithmetic operators (numeric).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ArithOp {
    Add,
    Sub,
    Mul,
    Div,
}

/// Supported SPARQL built-in functions (the unambiguous subset over our
/// term-token model).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Builtin {
    Str,
    StrLen,
    UCase,
    LCase,
    Abs,
    Ceil,
    Floor,
    Round,
    Concat,
    SubStr,
    StrBefore,
    StrAfter,
    Contains,
    StrStarts,
    StrEnds,
    IsIri,
    IsBlank,
    IsLiteral,
    IsNumeric,
    Datatype,
    Lang,
    Regex,
    LangMatches,
    StrDt,
    StrLang,
    Iri,
    EncodeForUri,
    Replace,
    Md5,
    Sha1,
    Sha256,
    Sha384,
    Sha512,
    Year,
    Month,
    Day,
    Hours,
    Minutes,
    Seconds,
    Timezone,
    Tz,
    CastInteger,
    CastDecimal,
    CastFloat,
    CastDouble,
    CastBoolean,
    CastString,
    Rand,
    Uuid,
    StrUuid,
    BNode,
    // GeoSPARQL (geof:) — topological relations → xsd:boolean, distance →
    // xsd:double, envelope → geo:wktLiteral. See crate::geo.
    GeoSfContains,
    GeoSfWithin,
    GeoSfIntersects,
    GeoSfDisjoint,
    GeoSfEquals,
    GeoDistance,
    GeoEnvelope,
    // geo3 — 3D extension of GeoSPARQL (AABB topology + distance). See crate::geo3.
    Geo3Distance,
    Geo3Contains,
    Geo3Within,
    Geo3Adjacent,
    // RDF-star / SPARQL-star: construct + inspect quoted triples (`<<s p o>>`).
    /// `TRIPLE(s, p, o)` → the quoted triple term.
    TripleTerm,
    /// `isTRIPLE(t)` → whether `t` is a quoted triple (a boolean builtin).
    IsTriple,
    /// `SUBJECT(t)` / `PREDICATE(t)` / `OBJECT(t)` → the component of a quoted triple.
    Subject,
    Predicate,
    Object,
}

/// A small boolean/comparison expression for FILTER (a subset of SPARQL exprs).
#[derive(Debug, Clone)]
pub enum FExpr {
    Var(String),
    /// A constant term token (IRI/literal) or numeric literal's text.
    Const(String),
    /// Numeric arithmetic on two sub-expressions.
    Arith(ArithOp, Box<FExpr>, Box<FExpr>),
    /// A built-in function call.
    Func(Builtin, Vec<FExpr>),
    /// `COALESCE(...)` — first sub-expression that yields a value.
    Coalesce(Vec<FExpr>),
    /// `IF(cond, then, else)` — evaluate `cond` as a boolean and pick a branch.
    If(Box<FExpr>, Box<FExpr>, Box<FExpr>),
    /// `expr IN (a, b, …)` — true if `expr` value-equals any list member.
    In(Box<FExpr>, Vec<FExpr>),
    /// `sameTerm(a, b)` — strict term identity (no value coercion).
    SameTerm(Box<FExpr>, Box<FExpr>),
    Compare(Op, Box<FExpr>, Box<FExpr>),
    And(Box<FExpr>, Box<FExpr>),
    Or(Box<FExpr>, Box<FExpr>),
    Not(Box<FExpr>),
    Bound(String),
    /// `EXISTS { … }` — true if the sub-pattern has a solution compatible with
    /// the current binding. (`NOT EXISTS` is `Not(Exists(..))`.)
    Exists(Box<Plan>),
}

/// Memoizes each EXISTS sub-plan's solutions (plus a lazily-built semi-join
/// index) within one filter application.
type ExistsCache = std::collections::HashMap<*const Plan, ExistsEntry>;

/// A cached EXISTS sub-plan: its solutions and a semi-join index built on first
/// probe, so repeated probes are O(1) instead of O(sols) — turning FILTER (NOT)
/// EXISTS over a BGP from O(L×R) into O(L+R), like the MINUS anti-join.
struct ExistsEntry {
    /// The sub-plan's solutions; filled on first need (`loaded`).
    sols: Vec<crate::row::Row>,
    loaded: bool,
    probe: Option<ExistsProbe>,
    /// A fat BGP sub-plan probed per outer row instead of loaded (`None` when
    /// loading it whole is the cheaper plan).
    corr: Option<crate::bgp::CorrelatedProbe>,
}

/// The semi-join index: solution rows keyed by the slots they share with the
/// probing rows. A probe `b` satisfies EXISTS iff some solution is compatible
/// with it (agrees on every shared slot).
struct ExistsProbe {
    /// All slots bound by some solution (ascending).
    svars: Vec<usize>,
    /// The shared slots with the probing rows (ascending) — the index key.
    jvars: Vec<usize>,
    /// `jvars`-value tuples of the solutions bound on all of `jvars`.
    keys: std::collections::HashSet<Vec<crate::row::Val>>,
    /// Solutions missing a `jvars` slot (e.g. via a nested OPTIONAL): scanned.
    partial: Vec<crate::row::Row>,
}

/// Build the semi-join index for `sols`, keyed by the slots shared with the
/// probe row `b`.
fn build_exists_probe(b: &crate::row::Row, sols: &[crate::row::Row]) -> ExistsProbe {
    let mask = crate::row::bound_mask(sols, b.len());
    let svars: Vec<usize> = (0..b.len()).filter(|&i| mask[i]).collect();
    let jvars: Vec<usize> = svars.iter().copied().filter(|&i| b[i].is_some()).collect();
    let mut keys = std::collections::HashSet::new();
    let mut partial = Vec::new();
    for s in sols {
        match jvars
            .iter()
            .map(|&i| s[i].clone())
            .collect::<Option<Vec<crate::row::Val>>>()
        {
            Some(k) => {
                keys.insert(k);
            }
            None => partial.push(s.clone()),
        }
    }
    ExistsProbe {
        svars,
        jvars,
        keys,
        partial,
    }
}

/// Does `b` satisfy the cached EXISTS? Uses the keyed index when `b`'s shared
/// slots match the index's `jvars` (the common, homogeneous case); otherwise
/// falls back to scanning all solutions (exact semantics on irregular rows).
fn exists_matches(b: &crate::row::Row, entry: &ExistsEntry) -> bool {
    let probe = entry.probe.as_ref().unwrap();
    let bj: Vec<usize> = probe
        .svars
        .iter()
        .copied()
        .filter(|&i| b[i].is_some())
        .collect();
    if bj == probe.jvars {
        let k: Vec<crate::row::Val> = probe.jvars.iter().map(|&i| b[i].clone().unwrap()).collect();
        probe.keys.contains(&k)
            || probe
                .partial
                .iter()
                .any(|s| crate::row::compatible_rows(b, s))
    } else {
        entry.sols.iter().any(|s| crate::row::compatible_rows(b, s))
    }
}

/// Lexical value of a term token: a literal's body with its N-Triples escapes
/// resolved (the `@lang` / `^^<dt>` suffix dropped), else the IRI/blank-node
/// token unchanged. The literal branch defers to [`crate::terms`] so that an
/// escaped `\"` inside the body is part of the value — a naive scan to the
/// first `"` cut `"He said \"hi\" loudly"` down to `He said \`, so CONTAINS /
/// REGEX / STRSTARTS / STRENDS / LANGMATCHES / GROUP_CONCAT never saw the rest.
fn lexical(token: &str) -> String {
    crate::terms::literal_lexical(token).unwrap_or_else(|| token.to_string())
}

/// Numeric value of a term for SUM / AVG: a literal of a numeric datatype with
/// a valid lexical form, else `None` — `"30"^^xsd:int` is 30, the string
/// `"30"` is not a number (§18.5.1.3 folds `op:numeric-add`). The crate-visible
/// name used by the row resolver's memoized parse.
pub(crate) fn term_number(s: &str) -> Option<f64> {
    value::numeric(s)
}

use crate::terms::as_number;

/// The result of evaluating any SPARQL query form.
#[derive(Debug, Clone)]
#[must_use]
#[non_exhaustive]
pub enum QueryOutput {
    /// SELECT: projected variables and their solution rows.
    Select(Vec<String>, Vec<Binding>),
    /// ASK: whether the pattern has any solution.
    Ask(bool),
    /// CONSTRUCT: the constructed triples as `(s, p, o)` term tokens.
    Construct(Vec<(String, String, String)>),
}

/// Query shapes that can be answered exactly from [`crate::range::SummaryView`] predicate
/// totals, without opening the triple index.
#[derive(Debug, Clone, PartialEq, Eq)]
#[must_use]
#[non_exhaustive]
pub enum SummaryQueryShape {
    /// `SELECT (COUNT(*) AS ?n) WHERE { ?s <p> ?o }`
    PredicateCount { predicate: String, variable: String },
    /// `SELECT (COUNT(*) AS ?n) WHERE { ?s ?p ?o }`
    TripleCount { variable: String },
    /// `SELECT ?p (COUNT(*) AS ?n) WHERE { ?s ?p ?o } GROUP BY ?p`
    PredicateTotals {
        predicate_variable: String,
        count_variable: String,
    },
    /// `SELECT DISTINCT ?p WHERE { ?s ?p ?o }`
    PredicateList { variable: String },
    /// `SELECT (COUNT(DISTINCT ?p) AS ?n) WHERE { ?s ?p ?o }`
    PredicateDistinctCount { variable: String },
    /// `ASK { ?s ?p ?o }`
    TripleExists,
    /// `ASK { ?s <p> ?o }`
    PredicateExists { predicate: String },
}

/// A single triple pattern that can be answered by the range-routed permutation
/// reader. `None` means the position is a variable/wildcard; `Some(term)` means
/// the query pins that term.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RoutedTriplePattern {
    pub subject: Option<String>,
    pub predicate: Option<String>,
    pub object: Option<String>,
}

/// Classify queries whose graph access is exactly one default-graph triple
/// pattern. Solution modifiers (projection, LIMIT, aggregate wrappers) do not
/// change the underlying range access, but named graphs, FROM, joins, filters,
/// paths, and other algebra need the full SPARQL evaluator.
pub fn routed_triple_pattern(query: &str) -> Result<Option<RoutedTriplePattern>, SparqlError> {
    let parsed = parse_query(query)?;
    let sel = match parsed {
        Query::Select {
            pattern, dataset, ..
        } => lower_select(&pattern, &dataset)?,
        Query::Ask { pattern, .. } => lower_pattern(&pattern)?,
        Query::Construct { pattern, .. } => lower_pattern(&pattern)?,
        Query::Describe { pattern, .. } => lower_pattern(&pattern)?,
    };
    if !sel.from.is_empty() || sel.from_named.is_some() {
        return Ok(None);
    }
    let Plan::Bgp(patterns) = sel.plan else {
        return Ok(None);
    };
    let [tp] = patterns.as_slice() else {
        return Ok(None);
    };
    Ok(Some(RoutedTriplePattern {
        subject: term_const(&tp.s),
        predicate: term_const(&tp.p),
        object: term_const(&tp.o),
    }))
}

fn term_const(term: &PatternTerm) -> Option<String> {
    match term {
        PatternTerm::Const(t) => Some(t.clone()),
        PatternTerm::Var(_) => None,
    }
}

/// Classify SPARQL queries that can be answered exactly from the pyramid
/// summary's per-predicate totals. This is intentionally conservative: anything
/// with constants, repeated variables, filters, joins, paths, named graphs,
/// ORDER BY, OFFSET/LIMIT, or non-summary-safe aggregates still requires the
/// index. The only accepted DISTINCT shape is a predicate list over one fully
/// unbound triple pattern.
pub fn summary_query_shape(query: &str) -> Result<Option<SummaryQueryShape>, SparqlError> {
    let parsed = parse_query(query)?;
    match parsed {
        Query::Select {
            pattern, dataset, ..
        } => {
            let sel = lower_select(&pattern, &dataset)?;
            if !sel.from.is_empty()
                || sel.from_named.is_some()
                || sel.offset != 0
                || sel.limit.is_some()
                || !sel.order.is_empty()
                || !sel.having.is_empty()
            {
                return Ok(None);
            }
            if sel.distinct {
                if sel.group.is_some() || !sel.extends.is_empty() {
                    return Ok(None);
                }
                let [projected] = sel.project.as_slice() else {
                    return Ok(None);
                };
                let Some(SummaryPatternShape::AnyPredicate { variable }) =
                    single_summary_pattern(&sel.plan)
                else {
                    return Ok(None);
                };
                return if projected == &variable {
                    Ok(Some(SummaryQueryShape::PredicateList { variable }))
                } else {
                    Ok(None)
                };
            }
            let Some(group) = &sel.group else {
                return Ok(None);
            };
            if group.aggs.len() != 1 {
                return Ok(None);
            }
            match group.by.as_slice() {
                [] => match &group.aggs[0].1 {
                    Agg::CountStar { distinct: false } => {
                        let Some(variable) = public_aggregate_variable(&sel, &group.aggs[0].0)
                        else {
                            return Ok(None);
                        };
                        Ok(single_summary_pattern(&sel.plan).map(|shape| match shape {
                            SummaryPatternShape::Predicate(predicate) => {
                                SummaryQueryShape::PredicateCount {
                                    predicate,
                                    variable,
                                }
                            }
                            SummaryPatternShape::AnyPredicate { .. } => {
                                SummaryQueryShape::TripleCount { variable }
                            }
                        }))
                    }
                    Agg::Count(counted, true) => {
                        let Some(public_variable) =
                            public_aggregate_variable(&sel, &group.aggs[0].0)
                        else {
                            return Ok(None);
                        };
                        let Some(SummaryPatternShape::AnyPredicate { variable }) =
                            single_summary_pattern(&sel.plan)
                        else {
                            return Ok(None);
                        };
                        if &variable != counted {
                            return Ok(None);
                        }
                        Ok(Some(SummaryQueryShape::PredicateDistinctCount {
                            variable: public_variable,
                        }))
                    }
                    _ => Ok(None),
                },
                [group_var] => {
                    if !matches!(group.aggs[0].1, Agg::CountStar { distinct: false }) {
                        return Ok(None);
                    }
                    let Some(SummaryPatternShape::AnyPredicate { variable }) =
                        single_summary_pattern(&sel.plan)
                    else {
                        return Ok(None);
                    };
                    if &variable != group_var {
                        return Ok(None);
                    }
                    let Some(count_variable) =
                        public_group_aggregate_variable(&sel, &group.aggs[0].0, group_var)
                    else {
                        return Ok(None);
                    };
                    Ok(Some(SummaryQueryShape::PredicateTotals {
                        predicate_variable: group_var.clone(),
                        count_variable,
                    }))
                }
                _ => Ok(None),
            }
        }
        Query::Ask { pattern, .. } => {
            let sel = lower_pattern(&pattern)?;
            Ok(single_summary_pattern(&sel.plan).map(|shape| match shape {
                SummaryPatternShape::Predicate(predicate) => {
                    SummaryQueryShape::PredicateExists { predicate }
                }
                SummaryPatternShape::AnyPredicate { .. } => SummaryQueryShape::TripleExists,
            }))
        }
        Query::Construct { .. } | Query::Describe { .. } => Ok(None),
    }
}

enum SummaryPatternShape {
    Predicate(String),
    AnyPredicate { variable: String },
}

fn single_summary_pattern(plan: &Plan) -> Option<SummaryPatternShape> {
    let Plan::Bgp(patterns) = plan else {
        return None;
    };
    let [tp] = patterns.as_slice() else {
        return None;
    };
    let (PatternTerm::Var(s), PatternTerm::Var(o)) = (&tp.s, &tp.o) else {
        return None;
    };
    if s == o {
        return None;
    }
    match &tp.p {
        PatternTerm::Const(p) => Some(SummaryPatternShape::Predicate(p.clone())),
        PatternTerm::Var(p) if p != s && p != o => Some(SummaryPatternShape::AnyPredicate {
            variable: p.clone(),
        }),
        _ => None,
    }
}

fn public_aggregate_variable(sel: &Select, aggregate_var: &str) -> Option<String> {
    let [projected] = sel.project.as_slice() else {
        return None;
    };
    if projected == aggregate_var {
        return Some(projected.clone());
    }
    sel.extends.iter().find_map(|(var, expr)| match expr {
        FExpr::Var(source) if var == projected && source == aggregate_var => Some(var.clone()),
        _ => None,
    })
}

fn public_group_aggregate_variable(
    sel: &Select,
    aggregate_var: &str,
    group_var: &str,
) -> Option<String> {
    let [projected_group, projected_aggregate] = sel.project.as_slice() else {
        return None;
    };
    if projected_group != group_var {
        return None;
    }
    if projected_aggregate == aggregate_var {
        return Some(projected_aggregate.clone());
    }
    sel.extends.iter().find_map(|(var, expr)| match expr {
        FExpr::Var(source) if var == projected_aggregate && source == aggregate_var => {
            Some(var.clone())
        }
        _ => None,
    })
}

/// One community's contribution to a community-split evaluation: how many
/// member subjects it holds and how many solution rows it produced.
#[derive(Debug, Clone, Copy)]
pub struct CommunityPartial {
    pub community: usize,
    pub subjects: usize,
    pub rows: usize,
}

/// The outcome of a community-split SELECT: the projected variables, the
/// merged solution rows, and each community's contribution.
pub type CommunitySelect = (Vec<String>, Vec<Binding>, Vec<CommunityPartial>);

/// Evaluate a SELECT **per pyramid community**, then merge: each community's
/// subjects are pushed into the plan as a VALUES binding, the partial rows
/// are concatenated, and the solution modifiers (GROUP BY / ORDER BY / LIMIT
/// / DISTINCT) run once on the union — so the rows are identical to
/// [`eval_query`]'s answer. Sound only for subject-star queries over the
/// default graph (every triple pattern sharing one subject variable; FILTERs
/// allowed); anything else returns [`SparqlError::Unsupported`] rather than a
/// possibly-wrong split answer. `round` picks the dendrogram granularity
/// (`None` = the build's tile-budget round). Also returns each community's
/// subject and row counts for display.
pub fn eval_select_communities(
    rete: &Rete,
    query: &str,
    round: Option<usize>,
) -> Result<CommunitySelect, SparqlError> {
    let parsed = parse_query(query)?;
    let out = match parsed {
        Query::Select {
            pattern, dataset, ..
        } => run_select_communities(rete, &lower_select(&pattern, &dataset)?, round),
        _ => Err(SparqlError::Unsupported(
            "community-split evaluation supports SELECT queries only",
        )),
    };
    // See `eval_query`: a failed non-SILENT SERVICE call must become an error
    // (and must not linger to poison a later query on the same handle).
    match rete.take_service_error() {
        Some(e) => Err(SparqlError::Service(e)),
        None => out,
    }
}

/// Evaluate any supported SPARQL query form (SELECT / ASK / CONSTRUCT).
pub fn eval_query(rete: &Rete, query: &str) -> Result<QueryOutput, SparqlError> {
    eval_query_with(rete, query, QueryOpts::default())
}

/// Like [`eval_query`], but with **OWL 2 QL entailment** on: the lowered plan is
/// rewritten by the internal QL lowering pass so the answer includes ontology-entailed solutions
/// (Stage 1a: `rdfs:subClassOf`), computed over the raw data with no
/// materialization. Opt-in — a plain [`eval_query`] is byte-identical to before.
pub fn eval_query_reasoned(rete: &Rete, query: &str) -> Result<QueryOutput, SparqlError> {
    eval_query_with(
        rete,
        query,
        QueryOpts {
            reason: true,
            ..QueryOpts::default()
        },
    )
}

/// Evaluation options for [`eval_query_with`]. Every flag defaults to off, and
/// off reproduces [`eval_query`] byte-for-byte — the W3C conformance suite runs
/// with this default.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct QueryOpts {
    /// OWL 2 QL entailment by query rewriting — see [`eval_query_reasoned`].
    pub reason: bool,
    /// Union default graph (non-standard, opt-in): a pattern outside `GRAPH`
    /// matches the RDF merge of the default graph and every named graph, the
    /// mode Virtuoso / GraphDB / Jena TDB offer. A query with its own `FROM`
    /// keeps its `FROM` dataset; `GRAPH ?g` still enumerates the named graphs
    /// exactly as before. See [`Select::union_default`].
    pub union_default_graph: bool,
}

/// [`eval_query`] with explicit [`QueryOpts`] — the single entry point the
/// opt-in toggles (🧠 reasoning, union default graph) funnel through.
pub fn eval_query_with(
    rete: &Rete,
    query: &str,
    opts: QueryOpts,
) -> Result<QueryOutput, SparqlError> {
    // A fresh diagnostics scope: the expression type errors this query raises
    // are counted on the side (see `diag`); `eval_query_with_warnings` reads
    // them, plain callers just never look.
    diag::begin();
    let out = eval_query_inner(rete, query, opts);
    // A failed non-SILENT SERVICE call is recorded out-of-band (the row
    // pipeline is infallible, like lazy tile fetches) — surface it here so a
    // partial answer is never returned as if it were complete.
    match rete.take_service_error() {
        Some(e) => Err(SparqlError::Service(e)),
        None => out,
    }
}

/// [`eval_query_with`], plus the diagnostics the evaluation produced.
///
/// SPARQL turns an expression **type error** into `false` inside FILTER (and
/// into an unbound value in BIND / projection / ORDER BY): `CONTAINS(?iri, "x")`
/// or `STRSTARTS(?year, "15")` on an `xsd:integer` silently drop every row. The
/// result here is exactly what [`eval_query_with`] returns — nothing is coerced
/// — but each class of such error comes back as a [`QueryWarning`]: the
/// function, the argument position, what the argument was (IRI, number,
/// language-incompatible literal, unbound, invalid regex, …), a count, one
/// sample value and a hint (`wrap it in STR()`). When the result is empty and no
/// error explains it, a case-sensitivity [`WarningSeverity::Hint`] may be added
/// for a constant CONTAINS / STRSTARTS / STRENDS / REGEX (no `i` flag) needle.
///
/// Errors are recorded only where they happen, so a query that raises none
/// pays nothing beyond a per-query reset; work per error is bounded (a count
/// and one sample per `(function, position, kind)`).
pub fn eval_query_with_warnings(
    rete: &Rete,
    query: &str,
    opts: QueryOpts,
) -> Result<(QueryOutput, Vec<QueryWarning>), SparqlError> {
    let out = eval_query_with(rete, query, opts)?;
    let warnings = diag::finish(&out);
    Ok((out, warnings))
}

/// Plan running `query` as many smaller queries, split by the values of
/// `var` (`?w`, `$w` or `w`), so a long extraction can go in short chunks
/// that each finish quickly and can be retried or resumed one by one.
///
/// Refused — [`SparqlError::Partition`], saying why — when the chunks'
/// answers together would not be the query's answer: an ASK or DESCRIBE, an
/// ORDER BY, an aggregate not grouped by `var`, or a `var` some solution may
/// leave unbound. Otherwise the plan carries the **keys query**, which lists
/// the values to split (a superset: a value with no solution makes an empty
/// chunk), and the top-level OFFSET/LIMIT the chunks drop and the caller
/// applies once to the merged rows. Run each chunk with
/// [`eval_query_partitioned`].
///
/// Reads only the tile directory (to pick the cheapest pattern for the keys
/// query), never tile data.
pub fn partition_plan(rete: &Rete, query: &str, var: &str) -> Result<PartitionPlan, SparqlError> {
    partition::plan(rete, query, var)
}

/// [`eval_query_with_warnings`] for one chunk of a [`partition_plan`]: the
/// query restricted to the solutions whose `var` is one of `keys` (term
/// tokens as the keys query returned them, e.g. `<https://…>`), without its
/// top-level OFFSET/LIMIT. The keys are joined in as `VALUES` on the first
/// required pattern block that binds `var`, so each one is an index lookup.
///
/// The chunks' rows, taken in the keys' order, are the query's rows — except
/// that the caller applies the plan's OFFSET/LIMIT and, when the plan says
/// `dedupe`, drops repeated rows.
pub fn eval_query_partitioned(
    rete: &Rete,
    query: &str,
    opts: QueryOpts,
    var: &str,
    keys: &[String],
) -> Result<(QueryOutput, Vec<QueryWarning>), SparqlError> {
    diag::begin();
    let out = eval_partition_inner(rete, query, opts, partition::var_name(var), keys);
    let out = match rete.take_service_error() {
        Some(e) => Err(SparqlError::Service(e)),
        None => out,
    }?;
    let warnings = diag::finish(&out);
    Ok((out, warnings))
}

fn eval_partition_inner(
    rete: &Rete,
    query: &str,
    opts: QueryOpts,
    var: &str,
    keys: &[String],
) -> Result<QueryOutput, SparqlError> {
    let parsed = parse_query(query)?;
    partition::check(&parsed, var)?;
    // Seeded after reasoning, so the VALUES lands on the pattern block the
    // engine will actually scan.
    let chunk = |sel: Select| -> Select {
        let mut sel = maybe_reason(rete, apply_opts(sel, opts), opts.reason);
        sel.offset = 0;
        sel.limit = None;
        sel.plan = partition::seed(sel.plan, var, keys);
        sel
    };
    match parsed {
        Query::Select {
            pattern, dataset, ..
        } => {
            let (vars, rows) = run_select(rete, &chunk(lower_select(&pattern, &dataset)?));
            Ok(QueryOutput::Select(vars, rows))
        }
        Query::Construct {
            template, pattern, ..
        } => {
            let sel = chunk(lower_pattern(&pattern)?);
            let (ctx, sols) = raw_solutions(rete, &sel);
            Ok(QueryOutput::Construct(instantiate(&ctx, &template, &sols)))
        }
        // `check` refused every other form.
        _ => Err(SparqlError::Unsupported(
            "only SELECT and CONSTRUCT can be split",
        )),
    }
}

/// Apply OWL 2 QL plan rewriting when reasoning is on (no-op otherwise). Reads a
/// small TBox slice from `rete` to gate the rewrite (see [`ql`]).
fn maybe_reason(rete: &Rete, mut sel: Select, reason: bool) -> Select {
    if reason {
        let projected = sel.project.clone();
        sel.plan = ql::reason_rewrite(sel.plan, rete, &projected);
    }
    sel
}

/// Stamp the caller's union-default-graph choice onto the lowered top-level
/// `Select` (dataset clauses are top-level-only in the SPARQL grammar, so this
/// never needs to descend into subqueries — they evaluate against the same
/// active index).
fn apply_opts(mut sel: Select, opts: QueryOpts) -> Select {
    sel.union_default = opts.union_default_graph;
    sel
}

fn eval_query_inner(rete: &Rete, query: &str, opts: QueryOpts) -> Result<QueryOutput, SparqlError> {
    let reason = opts.reason;
    let parsed = parse_query(query)?;
    match parsed {
        Query::Select {
            pattern, dataset, ..
        } => {
            let (vars, rows) = run_select(
                rete,
                &maybe_reason(
                    rete,
                    apply_opts(lower_select(&pattern, &dataset)?, opts),
                    reason,
                ),
            );
            Ok(QueryOutput::Select(vars, rows))
        }
        Query::Ask { pattern, .. } => {
            let sel = maybe_reason(rete, apply_opts(lower_pattern(&pattern)?, opts), reason);
            Ok(QueryOutput::Ask(ask_solution(rete, &sel)))
        }
        Query::Construct {
            template, pattern, ..
        } => {
            let sel = maybe_reason(rete, apply_opts(lower_pattern(&pattern)?, opts), reason);
            let (ctx, sols) = raw_solutions(rete, &sel);
            Ok(QueryOutput::Construct(instantiate(&ctx, &template, &sols)))
        }
        Query::Describe {
            pattern, dataset, ..
        } => {
            // The projected variables' values are the resources to describe;
            // we return each one's outgoing triples (concise bounded description).
            let sel = maybe_reason(
                rete,
                apply_opts(lower_select(&pattern, &dataset)?, opts),
                reason,
            );
            let (ctx, rows) = raw_solutions(rete, &sel);
            let mut resources = std::collections::BTreeSet::new();
            for row in &rows {
                if sel.project.is_empty() {
                    for val in row.iter().flatten() {
                        if let Some(t) = ctx.resolver.str_of(val) {
                            resources.insert(t.to_string());
                        }
                    }
                } else {
                    for v in &sel.project {
                        if let Some(val) = ctx.slots.slot(v).and_then(|s| row[s].as_ref()) {
                            if let Some(t) = ctx.resolver.str_of(val) {
                                resources.insert(t.to_string());
                            }
                        }
                    }
                }
            }
            // Described from the query's active default graph, the same graph
            // its WHERE clause matched: the file's default graph, the `FROM`
            // graphs, or (union-default toggle, no FROM) the default graph plus
            // every named graph.
            let mut graphs: Vec<Option<&str>> = Vec::new();
            if !sel.from.is_empty() {
                graphs.extend(sel.from.iter().map(|g| Some(g.as_str())));
            } else {
                graphs.push(None);
                if sel.union_default {
                    graphs.extend(rete.graph_names().into_iter().map(Some));
                }
            }
            let mut triples = std::collections::BTreeSet::new();
            for r in &resources {
                for g in &graphs {
                    for t in rete.query_in_graph(*g, Some(r), None, None) {
                        triples.insert(t);
                    }
                }
            }
            Ok(QueryOutput::Construct(triples.into_iter().collect()))
        }
    }
}

/// Parse and evaluate a SELECT against a file, applying the plan then
/// projection, DISTINCT, OFFSET, and LIMIT. Returns `(projected_vars,
/// solutions)`.
pub fn eval_sparql(rete: &Rete, query: &str) -> Result<(Vec<String>, Vec<Binding>), SparqlError> {
    eval_sparql_opts(rete, query, false)
}

/// Like [`eval_sparql`], but with OWL 2 QL entailment on (see [`eval_query_reasoned`]).
pub fn eval_sparql_reasoned(
    rete: &Rete,
    query: &str,
) -> Result<(Vec<String>, Vec<Binding>), SparqlError> {
    eval_sparql_opts(rete, query, true)
}

fn eval_sparql_opts(
    rete: &Rete,
    query: &str,
    reason: bool,
) -> Result<(Vec<String>, Vec<Binding>), SparqlError> {
    let sel = maybe_reason(rete, parse_select(query)?, reason);
    let out = run_select(rete, &sel);
    // See `eval_query`: a failed non-SILENT SERVICE call must become an error.
    match rete.take_service_error() {
        Some(e) => Err(SparqlError::Service(e)),
        None => Ok(out),
    }
}

/// Format a computed number as an N-Triples *typed* literal so the result
/// serializer emits its datatype (SPARQL requires arithmetic/aggregates/numeric
/// functions to yield typed numerics, not bare strings). Whole values are
/// `xsd:integer`; fractional ones `xsd:decimal` — the common cases in the data
/// (`xsd:double` would need operand-type tracking we don't carry through `f64`).
pub(crate) fn fmt_num_typed(x: f64) -> String {
    if x.fract() == 0.0 {
        format!(
            "\"{}\"^^<http://www.w3.org/2001/XMLSchema#integer>",
            x as i64
        )
    } else {
        // Round to 15 significant digits before emitting the shortest form, so a
        // sum/avg of decimals that lands on a binary-float artifact (e.g.
        // 11.100000000000001) serializes as the intended "11.1".
        let cleaned: f64 = format!("{x:.14e}").parse().unwrap_or(x);
        format!("\"{cleaned}\"^^<http://www.w3.org/2001/XMLSchema#decimal>")
    }
}

/// Push a reverse through a path (reverses each predicate and swaps sequences).
fn reverse(ast: PathAst) -> PathAst {
    match ast {
        PathAst::Pred(p, r) => PathAst::Pred(p, !r),
        PathAst::Rep(inner, rep) => PathAst::Rep(Box::new(reverse(*inner)), rep),
        PathAst::Seq(a, b) => PathAst::Seq(Box::new(reverse(*b)), Box::new(reverse(*a))),
        PathAst::Alt(a, b) => PathAst::Alt(Box::new(reverse(*a)), Box::new(reverse(*b))),
        PathAst::NegatedSet(s, r) => PathAst::NegatedSet(s, !r),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::dictionary::DictionaryBuilder;
    use crate::index::GraphIndexBuilder;
    use crate::write_file_v05 as write_file;

    fn rete_from(triples: &[(&str, &str, &str)]) -> Vec<u8> {
        let mut db = DictionaryBuilder::new();
        for (s, p, o) in triples {
            db.observe(s, p, o);
        }
        let dict = db.build();
        let mut ib = GraphIndexBuilder::new();
        for (s, p, o) in triples {
            ib.push(dict.encode(s, p, o).unwrap());
        }
        write_file(&dict, &ib.build(), false, &[], 0)
    }

    #[test]
    fn parses_select_with_two_patterns() {
        let q = r#"
            PREFIX ex: <http://ex/>
            SELECT ?x ?z WHERE { ?x ex:knows ?y . ?y ex:knows ?z }
        "#;
        let sel = parse_select(q).unwrap();
        assert_eq!(sel.project, vec!["x", "z"]);
        match &sel.plan {
            Plan::Bgp(p) => assert_eq!(p.len(), 2),
            other => panic!("expected a BGP plan, got {other:?}"),
        }
    }

    #[test]
    fn evaluates_two_hop_select() {
        let bytes = rete_from(&[
            ("<http://ex/Alice>", "<http://ex/knows>", "<http://ex/Bob>"),
            ("<http://ex/Bob>", "<http://ex/knows>", "<http://ex/Carol>"),
        ]);
        let rete = Rete::open(&bytes).unwrap();
        let q = r#"PREFIX ex: <http://ex/>
                   SELECT ?x ?z WHERE { ?x ex:knows ?y . ?y ex:knows ?z }"#;
        let (proj, sols) = eval_sparql(&rete, q).unwrap();
        assert_eq!(proj, vec!["x", "z"]);
        assert_eq!(sols.len(), 1);
        assert_eq!(sols[0]["x"], "<http://ex/Alice>");
        assert_eq!(sols[0]["z"], "<http://ex/Carol>");
    }

    #[test]
    fn union_returns_both_sides() {
        let bytes = rete_from(&[
            ("<http://ex/Alice>", "<http://ex/likes>", "<http://ex/Tea>"),
            ("<http://ex/Bob>", "<http://ex/hates>", "<http://ex/Tea>"),
        ]);
        let rete = Rete::open(&bytes).unwrap();
        let q = "PREFIX ex: <http://ex/> SELECT ?p WHERE { \
                 { ?p ex:likes ex:Tea } UNION { ?p ex:hates ex:Tea } }";
        let (_, sols) = eval_sparql(&rete, q).unwrap();
        let mut who: Vec<&str> = sols.iter().map(|b| b["p"].as_str()).collect();
        who.sort();
        assert_eq!(who, vec!["<http://ex/Alice>", "<http://ex/Bob>"]);
    }

    #[test]
    fn optional_keeps_left_when_right_absent() {
        // Alice has an email, Bob doesn't. OPTIONAL email keeps both people.
        let bytes = rete_from(&[
            ("<http://ex/Alice>", "<http://ex/name>", "\"Alice\""),
            ("<http://ex/Bob>", "<http://ex/name>", "\"Bob\""),
            ("<http://ex/Alice>", "<http://ex/email>", "\"a@ex\""),
        ]);
        let rete = Rete::open(&bytes).unwrap();
        let q = "PREFIX ex: <http://ex/> SELECT ?p ?e WHERE { \
                 ?p ex:name ?n . OPTIONAL { ?p ex:email ?e } }";
        let (_, sols) = eval_sparql(&rete, q).unwrap();
        assert_eq!(sols.len(), 2, "both people present");
        let alice = sols.iter().find(|b| b["p"] == "<http://ex/Alice>").unwrap();
        assert_eq!(alice["e"], "\"a@ex\"");
        let bob = sols.iter().find(|b| b["p"] == "<http://ex/Bob>").unwrap();
        assert!(!bob.contains_key("e"), "Bob has no email binding");
    }

    #[test]
    fn numeric_filter_on_typed_literal() {
        // ages 30 and 25; FILTER(?age > 27) keeps only Alice.
        let xsd = "<http://www.w3.org/2001/XMLSchema#integer>";
        let bytes = rete_from(&[
            (
                "<http://ex/Alice>",
                "<http://ex/age>",
                &format!("\"30\"^^{xsd}"),
            ),
            (
                "<http://ex/Bob>",
                "<http://ex/age>",
                &format!("\"25\"^^{xsd}"),
            ),
        ]);
        let rete = Rete::open(&bytes).unwrap();
        let q = "PREFIX ex: <http://ex/> \
                 SELECT ?p WHERE { ?p ex:age ?age . FILTER(?age > 27) }";
        let (_, sols) = eval_sparql(&rete, q).unwrap();
        assert_eq!(sols.len(), 1);
        assert_eq!(sols[0]["p"], "<http://ex/Alice>");
    }

    #[test]
    fn filter_equality_and_boolean_logic() {
        let bytes = rete_from(&[
            ("<http://ex/Alice>", "<http://ex/city>", "<http://ex/NYC>"),
            ("<http://ex/Bob>", "<http://ex/city>", "<http://ex/LA>"),
            ("<http://ex/Carol>", "<http://ex/city>", "<http://ex/NYC>"),
        ]);
        let rete = Rete::open(&bytes).unwrap();
        let q = "PREFIX ex: <http://ex/> \
                 SELECT ?p WHERE { ?p ex:city ?c . FILTER(?c = ex:NYC && ?p != ex:Carol) }";
        let (_, sols) = eval_sparql(&rete, q).unwrap();
        assert_eq!(sols.len(), 1);
        assert_eq!(sols[0]["p"], "<http://ex/Alice>");
    }

    #[test]
    fn distinct_collapses_duplicate_projections() {
        // Dave is reachable in two hops via both Bob and Carol.
        let bytes = rete_from(&[
            ("<http://ex/Alice>", "<http://ex/knows>", "<http://ex/Bob>"),
            (
                "<http://ex/Alice>",
                "<http://ex/knows>",
                "<http://ex/Carol>",
            ),
            ("<http://ex/Bob>", "<http://ex/knows>", "<http://ex/Dave>"),
            ("<http://ex/Carol>", "<http://ex/knows>", "<http://ex/Dave>"),
        ]);
        let rete = Rete::open(&bytes).unwrap();
        let base = "PREFIX ex: <http://ex/> SELECT {} ?z WHERE { ?x ex:knows ?y . ?y ex:knows ?z }";

        let (_, non_distinct) = eval_sparql(&rete, &base.replace("{}", "")).unwrap();
        assert_eq!(non_distinct.len(), 2); // {z:Dave} twice

        let (proj, distinct) = eval_sparql(&rete, &base.replace("{}", "DISTINCT")).unwrap();
        assert_eq!(proj, vec!["z"]);
        assert_eq!(distinct.len(), 1);
        assert_eq!(distinct[0]["z"], "<http://ex/Dave>");
        // Projection dropped ?x and ?y.
        assert!(!distinct[0].contains_key("x"));
    }

    #[test]
    fn transitive_path_reachability() {
        // A -> B -> C -> D chain.
        let bytes = rete_from(&[
            ("<http://ex/A>", "<http://ex/k>", "<http://ex/B>"),
            ("<http://ex/B>", "<http://ex/k>", "<http://ex/C>"),
            ("<http://ex/C>", "<http://ex/k>", "<http://ex/D>"),
        ]);
        let rete = Rete::open(&bytes).unwrap();

        // A k+ ?y  → B, C, D (one or more hops).
        let q = "PREFIX ex: <http://ex/> SELECT ?y WHERE { ex:A ex:k+ ?y }";
        let (_, sols) = eval_sparql(&rete, q).unwrap();
        let mut ys: Vec<&str> = sols.iter().map(|b| b["y"].as_str()).collect();
        ys.sort();
        assert_eq!(ys, vec!["<http://ex/B>", "<http://ex/C>", "<http://ex/D>"]);

        // A k* ?y includes A itself (zero-length).
        let q0 = "PREFIX ex: <http://ex/> SELECT ?y WHERE { ex:A ex:k* ?y }";
        let (_, s0) = eval_sparql(&rete, q0).unwrap();
        assert!(s0.iter().any(|b| b["y"] == "<http://ex/A>"));
        assert_eq!(s0.len(), 4); // A, B, C, D
    }

    #[test]
    fn sequence_and_alternative_paths() {
        // Alice -parent-> Bob -parent-> Carol; Alice -stepparent-> Dave.
        let bytes = rete_from(&[
            ("<http://ex/Alice>", "<http://ex/parent>", "<http://ex/Bob>"),
            ("<http://ex/Bob>", "<http://ex/parent>", "<http://ex/Carol>"),
            (
                "<http://ex/Alice>",
                "<http://ex/stepparent>",
                "<http://ex/Dave>",
            ),
        ]);
        let rete = Rete::open(&bytes).unwrap();

        // grandparent = parent/parent : Alice -> Carol.
        let q = "PREFIX ex: <http://ex/> SELECT ?g WHERE { ex:Alice ex:parent/ex:parent ?g }";
        let (_, sols) = eval_sparql(&rete, q).unwrap();
        assert_eq!(sols.len(), 1);
        assert_eq!(sols[0]["g"], "<http://ex/Carol>");

        // any-parent = parent|stepparent : Alice -> {Bob, Dave}.
        let qa = "PREFIX ex: <http://ex/> \
                  SELECT ?p WHERE { ex:Alice ex:parent|ex:stepparent ?p }";
        let (_, sa) = eval_sparql(&rete, qa).unwrap();
        let mut ps: Vec<&str> = sa.iter().map(|b| b["p"].as_str()).collect();
        ps.sort();
        assert_eq!(ps, vec!["<http://ex/Bob>", "<http://ex/Dave>"]);
    }

    #[test]
    fn group_concat_aggregate() {
        let bytes = rete_from(&[
            ("<http://ex/Alice>", "<http://ex/knows>", "<http://ex/Bob>"),
            (
                "<http://ex/Alice>",
                "<http://ex/knows>",
                "<http://ex/Carol>",
            ),
        ]);
        let rete = Rete::open(&bytes).unwrap();
        let q = "PREFIX ex: <http://ex/> \
                 SELECT (GROUP_CONCAT(?f; SEPARATOR=\"|\") AS ?fs) WHERE { ex:Alice ex:knows ?f }";
        let (_, sols) = eval_sparql(&rete, q).unwrap();
        assert_eq!(sols.len(), 1);
        // GROUP_CONCAT yields a simple literal — strip the quotes before splitting.
        let fs = sols[0]["fs"].trim_matches('"');
        let mut parts: Vec<&str> = fs.split('|').collect();
        parts.sort();
        assert_eq!(parts, vec!["<http://ex/Bob>", "<http://ex/Carol>"]);
    }

    #[test]
    fn group_by_having() {
        // Alice knows 2, Bob knows 1.
        let bytes = rete_from(&[
            ("<http://ex/Alice>", "<http://ex/knows>", "<http://ex/Bob>"),
            (
                "<http://ex/Alice>",
                "<http://ex/knows>",
                "<http://ex/Carol>",
            ),
            ("<http://ex/Bob>", "<http://ex/knows>", "<http://ex/Carol>"),
        ]);
        let rete = Rete::open(&bytes).unwrap();
        // Only people who know more than one person → Alice.
        let q = "PREFIX ex: <http://ex/> SELECT ?p (COUNT(?f) AS ?n) \
                 WHERE { ?p ex:knows ?f } GROUP BY ?p HAVING (COUNT(?f) > 1)";
        let (_, sols) = eval_sparql(&rete, q).unwrap();
        assert_eq!(sols.len(), 1);
        assert_eq!(sols[0]["p"], "<http://ex/Alice>");
        assert_eq!(
            sols[0]["n"],
            "\"2\"^^<http://www.w3.org/2001/XMLSchema#integer>"
        );
    }

    #[test]
    fn count_group_by() {
        // Alice knows Bob & Carol (2); Bob knows Carol (1).
        let bytes = rete_from(&[
            ("<http://ex/Alice>", "<http://ex/knows>", "<http://ex/Bob>"),
            (
                "<http://ex/Alice>",
                "<http://ex/knows>",
                "<http://ex/Carol>",
            ),
            ("<http://ex/Bob>", "<http://ex/knows>", "<http://ex/Carol>"),
        ]);
        let rete = Rete::open(&bytes).unwrap();
        let q = "PREFIX ex: <http://ex/> \
                 SELECT ?p (COUNT(?f) AS ?n) WHERE { ?p ex:knows ?f } GROUP BY ?p";
        let (_, sols) = eval_sparql(&rete, q).unwrap();
        let mut counts: Vec<(String, String)> = sols
            .iter()
            .map(|b| (b["p"].clone(), b["n"].clone()))
            .collect();
        counts.sort();
        assert_eq!(
            counts,
            vec![
                (
                    "<http://ex/Alice>".into(),
                    "\"2\"^^<http://www.w3.org/2001/XMLSchema#integer>".into()
                ),
                (
                    "<http://ex/Bob>".into(),
                    "\"1\"^^<http://www.w3.org/2001/XMLSchema#integer>".into()
                ),
            ]
        );
    }

    #[test]
    fn global_count_star() {
        let bytes = rete_from(&[
            ("<http://ex/a>", "<http://ex/p>", "<http://ex/1>"),
            ("<http://ex/b>", "<http://ex/p>", "<http://ex/2>"),
        ]);
        let rete = Rete::open(&bytes).unwrap();
        let (_, sols) = eval_sparql(&rete, "SELECT (COUNT(*) AS ?n) WHERE { ?s ?p ?o }").unwrap();
        assert_eq!(sols.len(), 1);
        assert_eq!(
            sols[0]["n"],
            "\"2\"^^<http://www.w3.org/2001/XMLSchema#integer>"
        );
    }

    #[test]
    fn summary_query_shape_classifies_only_exact_predicate_totals() {
        let count = summary_query_shape(
            "PREFIX ex: <http://ex/> SELECT (COUNT(*) AS ?n) WHERE { ?s ex:p ?o }",
        )
        .unwrap();
        assert_eq!(
            count,
            Some(SummaryQueryShape::PredicateCount {
                predicate: "<http://ex/p>".into(),
                variable: "n".into(),
            })
        );

        let total = summary_query_shape("SELECT (COUNT(*) AS ?n) WHERE { ?s ?p ?o }").unwrap();
        assert_eq!(
            total,
            Some(SummaryQueryShape::TripleCount {
                variable: "n".into(),
            })
        );

        let by_pred =
            summary_query_shape("SELECT ?p (COUNT(*) AS ?n) WHERE { ?s ?p ?o } GROUP BY ?p")
                .unwrap();
        assert_eq!(
            by_pred,
            Some(SummaryQueryShape::PredicateTotals {
                predicate_variable: "p".into(),
                count_variable: "n".into(),
            })
        );

        let predicates = summary_query_shape("SELECT DISTINCT ?p WHERE { ?s ?p ?o }").unwrap();
        assert_eq!(
            predicates,
            Some(SummaryQueryShape::PredicateList {
                variable: "p".into(),
            })
        );

        let predicate_count =
            summary_query_shape("SELECT (COUNT(DISTINCT ?p) AS ?n) WHERE { ?s ?p ?o }").unwrap();
        assert_eq!(
            predicate_count,
            Some(SummaryQueryShape::PredicateDistinctCount {
                variable: "n".into(),
            })
        );

        let ask = summary_query_shape("PREFIX ex: <http://ex/> ASK { ?s ex:p ?o }").unwrap();
        assert_eq!(
            ask,
            Some(SummaryQueryShape::PredicateExists {
                predicate: "<http://ex/p>".into(),
            })
        );

        let any_ask = summary_query_shape("ASK { ?s ?p ?o }").unwrap();
        assert_eq!(any_ask, Some(SummaryQueryShape::TripleExists));

        let constrained =
            summary_query_shape("PREFIX ex: <http://ex/> ASK { ex:a ex:p ?o }").unwrap();
        assert_eq!(constrained, None);

        let filtered =
            summary_query_shape("PREFIX ex: <http://ex/> ASK { ?s ex:p ?o FILTER(?s = ?o) }")
                .unwrap();
        assert_eq!(filtered, None);
    }

    #[test]
    fn filter_exists_and_not_exists() {
        // Alice knows Bob & Carol; Bob knows Dave; Carol knows nobody.
        let bytes = rete_from(&[
            ("<http://ex/Alice>", "<http://ex/knows>", "<http://ex/Bob>"),
            (
                "<http://ex/Alice>",
                "<http://ex/knows>",
                "<http://ex/Carol>",
            ),
            ("<http://ex/Bob>", "<http://ex/knows>", "<http://ex/Dave>"),
        ]);
        let rete = Rete::open(&bytes).unwrap();

        // Friends of Alice who themselves know someone → Bob.
        let q_exists = "PREFIX ex: <http://ex/> SELECT ?f WHERE { \
            ex:Alice ex:knows ?f . FILTER EXISTS { ?f ex:knows ?x } }";
        let (_, e) = eval_sparql(&rete, q_exists).unwrap();
        assert_eq!(
            e.iter().map(|b| b["f"].as_str()).collect::<Vec<_>>(),
            vec!["<http://ex/Bob>"]
        );

        // Friends of Alice who know nobody → Carol.
        let q_not = "PREFIX ex: <http://ex/> SELECT ?f WHERE { \
            ex:Alice ex:knows ?f . FILTER NOT EXISTS { ?f ex:knows ?x } }";
        let (_, n) = eval_sparql(&rete, q_not).unwrap();
        assert_eq!(
            n.iter().map(|b| b["f"].as_str()).collect::<Vec<_>>(),
            vec!["<http://ex/Carol>"]
        );
    }

    #[test]
    fn ask_repeated_variable_pattern() {
        // ASK's fast path must NOT take the single-pattern scan shortcut when a
        // variable repeats across positions (`?x knows ?x`) — only Bob self-knows.
        let yes = rete_from(&[
            ("<http://ex/Alice>", "<http://ex/knows>", "<http://ex/Bob>"),
            ("<http://ex/Bob>", "<http://ex/knows>", "<http://ex/Bob>"),
        ]);
        let rete = Rete::open(&yes).unwrap();
        match eval_query(&rete, "PREFIX ex: <http://ex/> ASK { ?x ex:knows ?x }").unwrap() {
            QueryOutput::Ask(b) => assert!(b, "Bob knows himself"),
            other => panic!("expected Ask, got {other:?}"),
        }
        // With no self-edge, the index still has a `knows` triple, so a naive
        // first-match probe would wrongly say true; the guard must reject it.
        let no = rete_from(&[("<http://ex/Alice>", "<http://ex/knows>", "<http://ex/Bob>")]);
        let rete = Rete::open(&no).unwrap();
        match eval_query(&rete, "PREFIX ex: <http://ex/> ASK { ?x ex:knows ?x }").unwrap() {
            QueryOutput::Ask(b) => assert!(!b, "nobody knows themselves"),
            other => panic!("expected Ask, got {other:?}"),
        }
    }

    #[test]
    fn minus_excludes_compatible() {
        // Alice knows Bob and Carol; Bob knows Carol; Carol knows no one.
        let bytes = rete_from(&[
            ("<http://ex/Alice>", "<http://ex/knows>", "<http://ex/Bob>"),
            (
                "<http://ex/Alice>",
                "<http://ex/knows>",
                "<http://ex/Carol>",
            ),
            ("<http://ex/Bob>", "<http://ex/knows>", "<http://ex/Carol>"),
        ]);
        let rete = Rete::open(&bytes).unwrap();
        // People Alice knows who themselves know nobody → Carol.
        let q = "PREFIX ex: <http://ex/> SELECT ?f WHERE { \
                 ex:Alice ex:knows ?f . MINUS { ?f ex:knows ?x } }";
        let (_, sols) = eval_sparql(&rete, q).unwrap();
        let fs: Vec<&str> = sols.iter().map(|b| b["f"].as_str()).collect();
        assert_eq!(fs, vec!["<http://ex/Carol>"]);
    }

    #[test]
    fn filter_exists_disjoint_variable() {
        // EXISTS over a sub-pattern that shares NO variable with the outer row is
        // true iff the sub-pattern has any solution — so NOT EXISTS removes ALL
        // rows (the exact case where NOT EXISTS differs from MINUS). Guards the
        // semi-join index's empty-`jvars` path.
        let bytes = rete_from(&[
            ("<http://ex/Alice>", "<http://ex/a>", "<http://ex/Person>"),
            ("<http://ex/Bob>", "<http://ex/a>", "<http://ex/Person>"),
            ("<http://ex/Tea>", "<http://ex/a>", "<http://ex/Drink>"),
        ]);
        let rete = Rete::open(&bytes).unwrap();
        let q_none = "PREFIX ex: <http://ex/> SELECT ?x WHERE { \
            ?x ex:a ex:Person FILTER NOT EXISTS { ?y ex:a ex:Drink } }";
        assert!(eval_sparql(&rete, q_none).unwrap().1.is_empty());
        let q_all = "PREFIX ex: <http://ex/> SELECT ?x WHERE { \
            ?x ex:a ex:Person FILTER EXISTS { ?y ex:a ex:Drink } }";
        assert_eq!(eval_sparql(&rete, q_all).unwrap().1.len(), 2);
    }

    #[test]
    fn minus_disjoint_domain_keeps_all() {
        // MINUS with no shared variable must remove nothing (SPARQL semantics) —
        // the hash anti-join's `jv.is_empty()` guard. Alice/Bob both kept even
        // though the right pattern has solutions.
        let bytes = rete_from(&[
            ("<http://ex/Alice>", "<http://ex/a>", "<http://ex/Person>"),
            ("<http://ex/Bob>", "<http://ex/a>", "<http://ex/Person>"),
            ("<http://ex/Tea>", "<http://ex/a>", "<http://ex/Drink>"),
        ]);
        let rete = Rete::open(&bytes).unwrap();
        let q = "PREFIX ex: <http://ex/> SELECT ?x WHERE { \
                 ?x ex:a ex:Person MINUS { ?y ex:a ex:Drink } }";
        let (_, sols) = eval_sparql(&rete, q).unwrap();
        let mut xs: Vec<&str> = sols.iter().map(|b| b["x"].as_str()).collect();
        xs.sort();
        assert_eq!(xs, vec!["<http://ex/Alice>", "<http://ex/Bob>"]);
    }

    #[test]
    fn values_pushdown_selects_subset() {
        // VALUES with several rows pushes each into the scan; the result must be
        // exactly the union (here: two of three disciplines).
        let bytes = rete_from(&[
            ("<http://ex/a>", "<http://ex/d>", "<http://ex/Bio>"),
            ("<http://ex/b>", "<http://ex/d>", "<http://ex/Phys>"),
            ("<http://ex/c>", "<http://ex/d>", "<http://ex/Chem>"),
        ]);
        let rete = Rete::open(&bytes).unwrap();
        let q = "PREFIX ex: <http://ex/> SELECT ?p ?disc WHERE { \
            VALUES ?disc { ex:Bio ex:Phys } ?p ex:d ?disc }";
        let (_, sols) = eval_sparql(&rete, q).unwrap();
        let mut got: Vec<(String, String)> = sols
            .iter()
            .map(|b| (b["p"].clone(), b["disc"].clone()))
            .collect();
        got.sort();
        assert_eq!(
            got,
            vec![
                ("<http://ex/a>".into(), "<http://ex/Bio>".into()),
                ("<http://ex/b>".into(), "<http://ex/Phys>".into()),
            ]
        );
    }

    #[test]
    fn values_inline_data_joins() {
        let bytes = rete_from(&[
            ("<http://ex/Alice>", "<http://ex/knows>", "<http://ex/Bob>"),
            (
                "<http://ex/Alice>",
                "<http://ex/knows>",
                "<http://ex/Carol>",
            ),
            ("<http://ex/Dave>", "<http://ex/knows>", "<http://ex/Eve>"),
        ]);
        let rete = Rete::open(&bytes).unwrap();
        // Restrict to Alice's friends via VALUES.
        let q = "PREFIX ex: <http://ex/> \
                 SELECT ?f WHERE { VALUES ?p { ex:Alice } ?p ex:knows ?f }";
        let (_, sols) = eval_sparql(&rete, q).unwrap();
        let mut fs: Vec<&str> = sols.iter().map(|b| b["f"].as_str()).collect();
        fs.sort();
        assert_eq!(fs, vec!["<http://ex/Bob>", "<http://ex/Carol>"]);
    }

    #[test]
    fn graph_queries_over_named_graphs() {
        use crate::write_dataset_v05 as write_dataset;
        // Shared dict; social graph has knows edges, profile graph has ages.
        let mut db = DictionaryBuilder::new();
        let edges = [
            ("<http://ex/Alice>", "<http://ex/knows>", "<http://ex/Bob>"),
            ("<http://ex/Bob>", "<http://ex/age>", "\"25\""),
        ];
        for (s, p, o) in edges {
            db.observe(s, p, o);
        }
        let dict = db.build();
        let mut social = GraphIndexBuilder::new();
        social.push(dict.encode(edges[0].0, edges[0].1, edges[0].2).unwrap());
        let mut profile = GraphIndexBuilder::new();
        profile.push(dict.encode(edges[1].0, edges[1].1, edges[1].2).unwrap());
        let named = vec![
            ("<http://ex/social>".to_string(), social.build()),
            ("<http://ex/profile>".to_string(), profile.build()),
        ];
        let bytes = write_dataset(
            &dict,
            &GraphIndexBuilder::new().build(),
            &named,
            true,
            &[],
            0,
        );
        let rete = Rete::open(&bytes).unwrap();

        // GRAPH <iri>: knows edge only in the social graph.
        let q = "PREFIX ex: <http://ex/> \
                 SELECT ?f WHERE { GRAPH ex:social { ex:Alice ex:knows ?f } }";
        let (_, s) = eval_sparql(&rete, q).unwrap();
        assert_eq!(s.len(), 1);
        assert_eq!(s[0]["f"], "<http://ex/Bob>");

        // GRAPH ?g: which graph holds the age triple?
        let q2 = "PREFIX ex: <http://ex/> \
                  SELECT ?g WHERE { GRAPH ?g { ex:Bob ex:age ?a } }";
        let (_, s2) = eval_sparql(&rete, q2).unwrap();
        assert_eq!(s2.len(), 1);
        assert_eq!(s2[0]["g"], "<http://ex/profile>");

        // EXISTS inside GRAPH evaluates in that graph: the `age` triple exists in
        // the profile graph but NOT in the social graph.
        let q3 = "PREFIX ex: <http://ex/> SELECT ?f WHERE { \
                  GRAPH ex:profile { ?f ex:age ?a . FILTER EXISTS { ?f ex:age ?a2 } } }";
        assert_eq!(eval_sparql(&rete, q3).unwrap().1.len(), 1);
        let q4 = "PREFIX ex: <http://ex/> SELECT ?s WHERE { \
                  GRAPH ex:social { ?s ex:knows ?o . FILTER NOT EXISTS { ?s ex:age ?a } } }";
        // In the social graph, no `age` triples exist → NOT EXISTS keeps the row.
        assert_eq!(eval_sparql(&rete, q4).unwrap().1.len(), 1);
    }

    #[test]
    fn from_unions_named_graphs() {
        use crate::write_dataset_v05 as write_dataset;
        // social graph: Alice knows Bob. profile graph: Bob knows Carol.
        // A join spanning both only works if FROM merges them into the default.
        let mut db = DictionaryBuilder::new();
        let t = [
            ("<http://ex/Alice>", "<http://ex/knows>", "<http://ex/Bob>"),
            ("<http://ex/Bob>", "<http://ex/knows>", "<http://ex/Carol>"),
        ];
        for (s, p, o) in t {
            db.observe(s, p, o);
        }
        let dict = db.build();
        let mut g1 = GraphIndexBuilder::new();
        g1.push(dict.encode(t[0].0, t[0].1, t[0].2).unwrap());
        let mut g2 = GraphIndexBuilder::new();
        g2.push(dict.encode(t[1].0, t[1].1, t[1].2).unwrap());
        let named = vec![
            ("<http://ex/social>".to_string(), g1.build()),
            ("<http://ex/profile>".to_string(), g2.build()),
        ];
        let bytes = write_dataset(
            &dict,
            &GraphIndexBuilder::new().build(),
            &named,
            true,
            &[],
            0,
        );
        let rete = Rete::open(&bytes).unwrap();

        // Without FROM the default graph is empty → no join.
        let q0 = "PREFIX ex: <http://ex/> SELECT ?z WHERE { ?x ex:knows ?y . ?y ex:knows ?z }";
        assert!(eval_sparql(&rete, q0).unwrap().1.is_empty());

        // FROM both graphs → the cross-graph join (Alice→Bob→Carol) succeeds.
        let q = "PREFIX ex: <http://ex/> \
                 SELECT ?z FROM ex:social FROM ex:profile \
                 WHERE { ?x ex:knows ?y . ?y ex:knows ?z }";
        let (_, s) = eval_sparql(&rete, q).unwrap();
        assert_eq!(s.len(), 1);
        assert_eq!(s[0]["z"], "<http://ex/Carol>");

        // FROM NAMED restricts GRAPH ?g to the listed graph only.
        let qn = "PREFIX ex: <http://ex/> SELECT ?g FROM NAMED ex:social \
                  WHERE { GRAPH ?g { ?x ex:knows ?y } }";
        let (_, sn) = eval_sparql(&rete, qn).unwrap();
        let gs: Vec<&str> = sn.iter().map(|b| b["g"].as_str()).collect();
        assert_eq!(gs, vec!["<http://ex/social>"]); // profile excluded
    }

    #[test]
    fn union_default_graph_opt_in() {
        use crate::write_dataset_v05 as write_dataset;
        // default graph: Root knows Alice. social: Alice knows Bob (twice — once
        // duplicated in profile, to pin set-union dedup). profile: the duplicate
        // plus Bob knows Carol.
        let mut db = DictionaryBuilder::new();
        let t = [
            ("<http://ex/Root>", "<http://ex/knows>", "<http://ex/Alice>"),
            ("<http://ex/Alice>", "<http://ex/knows>", "<http://ex/Bob>"),
            ("<http://ex/Bob>", "<http://ex/knows>", "<http://ex/Carol>"),
        ];
        for (s, p, o) in t {
            db.observe(s, p, o);
        }
        let dict = db.build();
        let enc = |i: usize| dict.encode(t[i].0, t[i].1, t[i].2).unwrap();
        let mut def = GraphIndexBuilder::new();
        def.push(enc(0));
        let mut g1 = GraphIndexBuilder::new();
        g1.push(enc(1));
        let mut g2 = GraphIndexBuilder::new();
        g2.push(enc(1)); // the SAME triple as social's — a set union keeps one
        g2.push(enc(2));
        let named = vec![
            ("<http://ex/social>".to_string(), g1.build()),
            ("<http://ex/profile>".to_string(), g2.build()),
        ];
        let bytes = write_dataset(&dict, &def.build(), &named, true, &[], 0);
        let rete = Rete::open(&bytes).unwrap();
        let union = QueryOpts {
            union_default_graph: true,
            ..QueryOpts::default()
        };
        let rows = |out: QueryOutput| match out {
            QueryOutput::Select(_, rows) => rows,
            other => panic!("expected Select, got {other:?}"),
        };

        // OFF (the default): a pattern outside GRAPH sees only the real default
        // graph — one triple. This is the standard-semantics baseline the W3C
        // conformance suite runs on.
        let q = "PREFIX ex: <http://ex/> SELECT ?s ?o WHERE { ?s ex:knows ?o }";
        assert_eq!(rows(eval_query(&rete, q).unwrap()).len(), 1);

        // ON: default ∪ social ∪ profile, deduplicated — 3 distinct triples,
        // not 4 (the duplicate collapses: RDF merge is a set union). The
        // default graph's own triple is INCLUDED (unlike a bare FROM).
        let got = rows(eval_query_with(&rete, q, union).unwrap());
        assert_eq!(got.len(), 3);
        assert!(got.iter().any(|b| b["s"] == "<http://ex/Root>"));

        // ON: a cross-graph join (default→social→profile) now succeeds.
        let qj = "PREFIX ex: <http://ex/> SELECT ?z WHERE { \
                  ex:Root ex:knows ?x . ?x ex:knows ?y . ?y ex:knows ?z }";
        let sj = rows(eval_query_with(&rete, qj, union).unwrap());
        assert_eq!(sj.len(), 1);
        assert_eq!(sj[0]["z"], "<http://ex/Carol>");

        // ON: GRAPH ?g still enumerates the named graphs exactly as before.
        let qg = "PREFIX ex: <http://ex/> SELECT DISTINCT ?g WHERE { GRAPH ?g { ?s ex:knows ?o } }";
        assert_eq!(rows(eval_query_with(&rete, qg, union).unwrap()).len(), 2);

        // ON + explicit FROM: the query's own dataset clause wins — FROM social
        // alone excludes both the default graph and profile.
        let qf = "PREFIX ex: <http://ex/> SELECT ?s ?o FROM ex:social WHERE { ?s ex:knows ?o }";
        let sf = rows(eval_query_with(&rete, qf, union).unwrap());
        assert_eq!(sf.len(), 1);
        assert_eq!(sf[0]["s"], "<http://ex/Alice>");

        // ON, empty-default single-named-graph shape (the zero-copy borrow arm):
        // build a second dataset with an empty default and ONE named graph.
        let mut db2 = DictionaryBuilder::new();
        db2.observe(t[1].0, t[1].1, t[1].2);
        let dict2 = db2.build();
        let mut only = GraphIndexBuilder::new();
        only.push(dict2.encode(t[1].0, t[1].1, t[1].2).unwrap());
        let named2 = vec![("<http://ex/only>".to_string(), only.build())];
        let bytes2 = write_dataset(
            &dict2,
            &GraphIndexBuilder::new().build(),
            &named2,
            true,
            &[],
            0,
        );
        let rete2 = Rete::open(&bytes2).unwrap();
        assert_eq!(rows(eval_query(&rete2, q).unwrap()).len(), 0); // standard: 0
        assert_eq!(rows(eval_query_with(&rete2, q, union).unwrap()).len(), 1);
    }

    #[test]
    fn describe_resource() {
        let bytes = rete_from(&[
            ("<http://ex/Alice>", "<http://ex/knows>", "<http://ex/Bob>"),
            ("<http://ex/Alice>", "<http://ex/age>", "\"30\""),
            ("<http://ex/Bob>", "<http://ex/age>", "\"25\""),
        ]);
        let rete = Rete::open(&bytes).unwrap();
        // DESCRIBE <Alice> → Alice's two outgoing triples.
        match eval_query(&rete, "DESCRIBE <http://ex/Alice>").unwrap() {
            QueryOutput::Construct(t) => assert_eq!(t.len(), 2),
            other => panic!("expected Construct, got {other:?}"),
        }
        // DESCRIBE ?x WHERE { ?x ex:age ?a } → describes Alice and Bob.
        let q = "PREFIX ex: <http://ex/> DESCRIBE ?x WHERE { ?x ex:age ?a }";
        match eval_query(&rete, q).unwrap() {
            QueryOutput::Construct(t) => {
                // Alice has 2 triples, Bob has 1 → 3 total.
                assert_eq!(t.len(), 3);
            }
            other => panic!("expected Construct, got {other:?}"),
        }
    }

    #[test]
    fn ask_and_construct() {
        let bytes = rete_from(&[
            ("<http://ex/Alice>", "<http://ex/knows>", "<http://ex/Bob>"),
            ("<http://ex/Bob>", "<http://ex/knows>", "<http://ex/Carol>"),
        ]);
        let rete = Rete::open(&bytes).unwrap();

        // ASK: is there any knows edge? yes; a likes edge? no.
        match eval_query(&rete, "PREFIX ex: <http://ex/> ASK { ?a ex:knows ?b }").unwrap() {
            QueryOutput::Ask(b) => assert!(b),
            other => panic!("expected Ask, got {other:?}"),
        }
        match eval_query(&rete, "PREFIX ex: <http://ex/> ASK { ?a ex:likes ?b }").unwrap() {
            QueryOutput::Ask(b) => assert!(!b),
            other => panic!("expected Ask, got {other:?}"),
        }

        // CONSTRUCT a reverse `knownBy` graph.
        let q = "PREFIX ex: <http://ex/> \
                 CONSTRUCT { ?b ex:knownBy ?a } WHERE { ?a ex:knows ?b }";
        match eval_query(&rete, q).unwrap() {
            QueryOutput::Construct(mut triples) => {
                triples.sort();
                assert_eq!(triples.len(), 2);
                assert!(triples.contains(&(
                    "<http://ex/Bob>".into(),
                    "<http://ex/knownBy>".into(),
                    "<http://ex/Alice>".into(),
                )));
            }
            other => panic!("expected Construct, got {other:?}"),
        }
    }

    #[test]
    fn substr_strbefore_strafter() {
        let bytes = rete_from(&[("<http://ex/a>", "<http://ex/name>", "\"Alice Smith\"")]);
        let rete = Rete::open(&bytes).unwrap();
        let q = "PREFIX ex: <http://ex/> SELECT ?first ?last ?ini WHERE { \
            ?p ex:name ?n . \
            BIND(STRBEFORE(?n, \" \") AS ?first) \
            BIND(STRAFTER(?n, \" \") AS ?last) \
            BIND(SUBSTR(?n, 1, 1) AS ?ini) }";
        let (_, sols) = eval_sparql(&rete, q).unwrap();
        // String built-ins return proper literal terms (quoted), not bare text.
        assert_eq!(sols[0]["first"], "\"Alice\"");
        assert_eq!(sols[0]["last"], "\"Smith\"");
        assert_eq!(sols[0]["ini"], "\"A\"");
    }

    #[test]
    fn concat_and_coalesce() {
        // Alice has a nickname, Bob doesn't.
        let bytes = rete_from(&[
            ("<http://ex/Alice>", "<http://ex/name>", "\"Alice\""),
            ("<http://ex/Alice>", "<http://ex/nick>", "\"Al\""),
            ("<http://ex/Bob>", "<http://ex/name>", "\"Bob\""),
        ]);
        let rete = Rete::open(&bytes).unwrap();
        // COALESCE falls back to name when nick is unbound; CONCAT builds a label.
        let q = "PREFIX ex: <http://ex/> SELECT ?label WHERE { \
            ?p ex:name ?name . OPTIONAL { ?p ex:nick ?nick } \
            BIND(CONCAT(\"@\", COALESCE(?nick, ?name)) AS ?label) }";
        let (_, sols) = eval_sparql(&rete, q).unwrap();
        let mut labels: Vec<&str> = sols.iter().map(|b| b["label"].as_str()).collect();
        labels.sort();
        assert_eq!(labels, vec!["\"@Al\"", "\"@Bob\""]);
    }

    #[test]
    fn builtin_functions() {
        let bytes = rete_from(&[
            ("<http://ex/Alice>", "<http://ex/name>", "\"Alice Smith\""),
            ("<http://ex/Bob>", "<http://ex/name>", "\"Bob Jones\""),
        ]);
        let rete = Rete::open(&bytes).unwrap();
        // CONTAINS on the literal value, and STRLEN as a computed value.
        let q = "PREFIX ex: <http://ex/> SELECT ?p ?len WHERE { \
            ?p ex:name ?n . FILTER(CONTAINS(?n, \"Smith\")) BIND(STRLEN(?n) AS ?len) }";
        let (_, sols) = eval_sparql(&rete, q).unwrap();
        assert_eq!(sols.len(), 1);
        assert_eq!(sols[0]["p"], "<http://ex/Alice>");
        assert_eq!(
            sols[0]["len"],
            "\"11\"^^<http://www.w3.org/2001/XMLSchema#integer>"
        ); // "Alice Smith"
    }

    #[test]
    fn geosparql_filter_and_functions() {
        let wkt = "\"POLYGON((0 0,10 0,10 10,0 10,0 0))\"^^\
            <http://www.opengis.net/ont/geosparql#wktLiteral>";
        let bytes = rete_from(&[
            (
                "<http://ex/f>",
                "<http://www.opengis.net/ont/geosparql#hasGeometry>",
                "<http://ex/f/g>",
            ),
            (
                "<http://ex/f/g>",
                "<http://www.opengis.net/ont/geosparql#asWKT>",
                wkt,
            ),
            (
                "<http://ex/f>",
                "<http://ex/year>",
                "\"1815\"^^<http://www.w3.org/2001/XMLSchema#integer>",
            ),
        ]);
        let rete = Rete::open(&bytes).unwrap();
        let pre = "PREFIX geo: <http://www.opengis.net/ont/geosparql#> \
            PREFIX geof: <http://www.opengis.net/def/function/geosparql/> \
            PREFIX uom: <http://www.opengis.net/def/uom/OGC/1.0/> PREFIX ex: <http://ex/> ";

        // Headline: temporal FILTER + spatial point-in-polygon compose.
        let (_, s) = eval_sparql(
            &rete,
            &format!(
                "{pre}SELECT ?f WHERE {{ ?f ex:year ?y ; \
            geo:hasGeometry/geo:asWKT ?w . \
            FILTER(?y = 1815 && geof:sfContains(?w, \"POINT(5 5)\"^^geo:wktLiteral)) }}"
            ),
        )
        .unwrap();
        assert_eq!(s.len(), 1, "point inside the polygon in year 1815");
        assert_eq!(s[0]["f"], "<http://ex/f>");

        // Point outside → no rows.
        let (_, s) = eval_sparql(
            &rete,
            &format!(
                "{pre}SELECT ?f WHERE {{ \
            ?f geo:hasGeometry/geo:asWKT ?w . \
            FILTER(geof:sfContains(?w, \"POINT(50 50)\"^^geo:wktLiteral)) }}"
            ),
        )
        .unwrap();
        assert!(s.is_empty());

        // sfWithin is the argument-swapped relation.
        let (_, s) = eval_sparql(
            &rete,
            &format!(
                "{pre}SELECT ?f WHERE {{ \
            ?f geo:hasGeometry/geo:asWKT ?w . \
            FILTER(geof:sfWithin(\"POINT(5 5)\"^^geo:wktLiteral, ?w)) }}"
            ),
        )
        .unwrap();
        assert_eq!(s.len(), 1);

        // Malformed WKT is a type error → 0 rows, but the query must NOT error.
        let (_, s) = eval_sparql(
            &rete,
            &format!(
                "{pre}SELECT ?f WHERE {{ \
            ?f geo:hasGeometry/geo:asWKT ?w . \
            FILTER(geof:sfContains(?w, \"garbage\"^^geo:wktLiteral)) }}"
            ),
        )
        .unwrap();
        assert!(s.is_empty());

        // Relation in BIND (value position) → typed xsd:boolean.
        let (_, s) = eval_sparql(
            &rete,
            &format!(
                "{pre}SELECT ?hit WHERE {{ \
            ?f geo:hasGeometry/geo:asWKT ?w . \
            BIND(geof:sfContains(?w, \"POINT(5 5)\"^^geo:wktLiteral) AS ?hit) }}"
            ),
        )
        .unwrap();
        assert_eq!(
            s[0]["hit"],
            "\"true\"^^<http://www.w3.org/2001/XMLSchema#boolean>"
        );

        // distance → xsd:double; envelope → geo:wktLiteral.
        let (_, s) = eval_sparql(
            &rete,
            &format!(
                "{pre}SELECT ?d WHERE {{ BIND(geof:distance(\
            \"POINT(0 0)\"^^geo:wktLiteral, \"POINT(0 1)\"^^geo:wktLiteral, uom:metre) AS ?d) }}"
            ),
        )
        .unwrap();
        assert!(
            s[0]["d"].ends_with("XMLSchema#double>"),
            "distance is xsd:double: {}",
            s[0]["d"]
        );
        let (_, s) = eval_sparql(
            &rete,
            &format!(
                "{pre}SELECT ?e WHERE {{ \
            ?f geo:hasGeometry/geo:asWKT ?w . BIND(geof:envelope(?w) AS ?e) }}"
            ),
        )
        .unwrap();
        assert!(s[0]["e"].contains("wktLiteral") && s[0]["e"].contains("POLYGON"));

        // An unsupported geof: function is rejected cleanly at parse/lower time.
        assert!(eval_sparql(
            &rete,
            &format!(
                "{pre}SELECT ?x WHERE {{ \
            BIND(geof:buffer(\"POINT(0 0)\"^^geo:wktLiteral, 1) AS ?x) }}"
            )
        )
        .is_err());
    }

    #[test]
    fn bind_arithmetic() {
        let xsd = "<http://www.w3.org/2001/XMLSchema#integer>";
        let bytes = rete_from(&[(
            "<http://ex/a>",
            "<http://ex/age>",
            &format!("\"30\"^^{xsd}"),
        )]);
        let rete = Rete::open(&bytes).unwrap();
        // BIND a computed value, and FILTER on arithmetic.
        let q = "PREFIX ex: <http://ex/> \
                 SELECT ?next WHERE { ?p ex:age ?age . BIND(?age + 1 AS ?next) FILTER(?age * 2 > 50) }";
        let (_, sols) = eval_sparql(&rete, q).unwrap();
        assert_eq!(sols.len(), 1);
        assert_eq!(
            sols[0]["next"],
            "\"31\"^^<http://www.w3.org/2001/XMLSchema#integer>"
        );
    }

    #[test]
    fn bind_value_is_visible_to_a_following_filter_and_join() {
        // A BIND inside the WHERE pattern must be evaluated before a *following*
        // FILTER (and join) can reference it — the bound var is in-tree, not a
        // projection-time alias.
        let xsd = "<http://www.w3.org/2001/XMLSchema#integer>";
        let n = |v: i32| format!("\"{v}\"^^{xsd}");
        let bytes = rete_from(&[
            ("<http://ex/a>", "<http://ex/v>", &n(1)),
            ("<http://ex/b>", "<http://ex/v>", &n(2)),
            ("<http://ex/c>", "<http://ex/v>", &n(3)),
            // a node whose :v equals b's value+1, for the join case
            ("<http://ex/x>", "<http://ex/v>", &n(3)),
        ]);
        let rete = Rete::open(&bytes).unwrap();

        // FILTER references the BIND'd ?z.
        let q1 = "PREFIX ex: <http://ex/> SELECT ?s WHERE { \
            ?s ex:v ?o . BIND(?o + 1 AS ?z) FILTER(?z = 3) }";
        let (_, s1) = eval_sparql(&rete, q1).unwrap();
        assert_eq!(s1.len(), 1, "only b (2+1=3) passes");
        assert_eq!(s1[0]["s"], "<http://ex/b>");

        // A following triple pattern joins on the BIND'd ?z.
        let q2 = "PREFIX ex: <http://ex/> SELECT ?s ?s2 WHERE { \
            ?s ex:v ?o . BIND(?o + 1 AS ?z) ?s2 ex:v ?z }";
        let (_, s2) = eval_sparql(&rete, q2).unwrap();
        // a (1→2) joins b (:v 2); b (2→3) joins c and x (:v 3); c (3→4) no match.
        let mut pairs: Vec<(String, String)> = s2
            .iter()
            .map(|b| (b["s"].clone(), b["s2"].clone()))
            .collect();
        pairs.sort();
        assert_eq!(
            pairs,
            vec![
                ("<http://ex/a>".to_string(), "<http://ex/b>".to_string()),
                ("<http://ex/b>".to_string(), "<http://ex/c>".to_string()),
                ("<http://ex/b>".to_string(), "<http://ex/x>".to_string()),
            ]
        );
    }

    #[test]
    fn order_by_numeric_desc_then_limit() {
        let xsd = "<http://www.w3.org/2001/XMLSchema#integer>";
        let bytes = rete_from(&[
            (
                "<http://ex/a>",
                "<http://ex/age>",
                &format!("\"30\"^^{xsd}"),
            ),
            (
                "<http://ex/b>",
                "<http://ex/age>",
                &format!("\"25\"^^{xsd}"),
            ),
            (
                "<http://ex/c>",
                "<http://ex/age>",
                &format!("\"40\"^^{xsd}"),
            ),
        ]);
        let rete = Rete::open(&bytes).unwrap();
        // Oldest two, descending by age.
        let q = "PREFIX ex: <http://ex/> \
                 SELECT ?p WHERE { ?p ex:age ?age } ORDER BY DESC(?age) LIMIT 2";
        let (_, sols) = eval_sparql(&rete, q).unwrap();
        let ps: Vec<&str> = sols.iter().map(|b| b["p"].as_str()).collect();
        assert_eq!(ps, vec!["<http://ex/c>", "<http://ex/a>"]); // 40, 30
    }

    #[test]
    fn limit_early_out_two_hop_join() {
        // A→B, B→C, B→D, C→E. Two-hop join (?x k ?y . ?y k ?z) has 3 solutions:
        // (A,B,C), (A,B,D), (B,C,E). The LIMIT early-out must preserve the count
        // contract and only ever yield genuine solutions.
        let bytes = rete_from(&[
            ("<http://ex/A>", "<http://ex/k>", "<http://ex/B>"),
            ("<http://ex/B>", "<http://ex/k>", "<http://ex/C>"),
            ("<http://ex/B>", "<http://ex/k>", "<http://ex/D>"),
            ("<http://ex/C>", "<http://ex/k>", "<http://ex/E>"),
        ]);
        let rete = Rete::open(&bytes).unwrap();
        let q = "PREFIX ex: <http://ex/> SELECT ?x ?z WHERE { ?x ex:k ?y . ?y ex:k ?z }";
        let (_, full) = eval_sparql(&rete, q).unwrap();
        assert_eq!(full.len(), 3);

        let (_, one) = eval_sparql(&rete, &format!("{q} LIMIT 1")).unwrap();
        assert_eq!(one.len(), 1);
        // The early-out row must be a real solution of the full query.
        assert!(one.iter().all(|r| full.contains(r)));

        // LIMIT above the total returns everything; OFFSET composes.
        let (_, all) = eval_sparql(&rete, &format!("{q} LIMIT 100")).unwrap();
        assert_eq!(all.len(), 3);
        let (_, off) = eval_sparql(&rete, &format!("{q} LIMIT 100 OFFSET 2")).unwrap();
        assert_eq!(off.len(), 1);
    }

    #[test]
    fn limit_early_out_filter_over_bgp() {
        // FILTER over a BGP under LIMIT streams and stops early; the result must
        // match the unlimited filtered query's prefix.
        let xsd = "<http://www.w3.org/2001/XMLSchema#integer>";
        let bytes = rete_from(&[
            ("<http://ex/a>", "<http://ex/n>", &format!("\"10\"^^{xsd}")),
            ("<http://ex/b>", "<http://ex/n>", &format!("\"20\"^^{xsd}")),
            ("<http://ex/c>", "<http://ex/n>", &format!("\"30\"^^{xsd}")),
            ("<http://ex/d>", "<http://ex/n>", &format!("\"40\"^^{xsd}")),
        ]);
        let rete = Rete::open(&bytes).unwrap();
        let q = "PREFIX ex: <http://ex/> SELECT ?p WHERE { ?p ex:n ?v FILTER(?v > 15) }";
        let (_, full) = eval_sparql(&rete, q).unwrap();
        assert_eq!(full.len(), 3); // b, c, d
        let (_, two) = eval_sparql(&rete, &format!("{q} LIMIT 2")).unwrap();
        assert_eq!(two.len(), 2);
        assert!(two.iter().all(|r| full.contains(r)));
    }

    #[test]
    fn distinct_bgp_fast_matches_general() {
        // a/b both →x, b→y, c→z. The integer-DISTINCT fast path must collapse on
        // the projection and apply OFFSET/LIMIT after dedup.
        let bytes = rete_from(&[
            ("<http://ex/a>", "<http://ex/p>", "<http://ex/x>"),
            ("<http://ex/b>", "<http://ex/p>", "<http://ex/x>"),
            ("<http://ex/b>", "<http://ex/p>", "<http://ex/y>"),
            ("<http://ex/c>", "<http://ex/p>", "<http://ex/z>"),
        ]);
        let rete = Rete::open(&bytes).unwrap();
        // DISTINCT ?o → {x, y, z} = 3 (the two ?s→x rows collapse).
        let q = "PREFIX ex: <http://ex/> SELECT DISTINCT ?o WHERE { ?s ex:p ?o }";
        let (proj, sols) = eval_sparql(&rete, q).unwrap();
        assert_eq!(proj, vec!["o"]);
        let mut os: Vec<&str> = sols.iter().map(|b| b["o"].as_str()).collect();
        os.sort();
        assert_eq!(os, vec!["<http://ex/x>", "<http://ex/y>", "<http://ex/z>"]);
        // LIMIT applies after dedup.
        assert_eq!(
            eval_sparql(&rete, &format!("{q} LIMIT 2")).unwrap().1.len(),
            2
        );
        // A 2-var DISTINCT keeps every (s, o) pair → 4 rows.
        let q2 = "PREFIX ex: <http://ex/> SELECT DISTINCT ?s ?o WHERE { ?s ex:p ?o }";
        assert_eq!(eval_sparql(&rete, q2).unwrap().1.len(), 4);
    }

    #[test]
    fn limit_caps_solutions() {
        let bytes = rete_from(&[
            ("<http://ex/a>", "<http://ex/p>", "<http://ex/1>"),
            ("<http://ex/b>", "<http://ex/p>", "<http://ex/2>"),
            ("<http://ex/c>", "<http://ex/p>", "<http://ex/3>"),
        ]);
        let rete = Rete::open(&bytes).unwrap();
        let (_, sols) =
            eval_sparql(&rete, "SELECT ?x WHERE { ?x <http://ex/p> ?y } LIMIT 2").unwrap();
        assert_eq!(sols.len(), 2);
    }

    /// End to end through the file and the memoizing resolver: a literal with an
    /// escaped quote in its value matches on the text AFTER the quote. The
    /// boolean string predicates used to read the stored token only up to its
    /// first `"`, so of these four labels CONTAINS("Quantum") found one instead
    /// of three and CONTAINS("Gases") none — while STRLEN, on the other code
    /// path, already saw all 25 characters.
    #[test]
    fn string_filters_see_past_an_embedded_quote() {
        let bytes = rete_from(&[
            (
                "<http://ex/a>",
                "<http://ex/label>",
                "\"plain Quantum here\"",
            ),
            (
                "<http://ex/b>",
                "<http://ex/label>",
                r#""He said \"Quantum\" loudly"@en"#,
            ),
            (
                "<http://ex/c>",
                "<http://ex/label>",
                r#""Theory of \"Quantum\" Gases""#,
            ),
            ("<http://ex/d>", "<http://ex/label>", "\"no match at all\""),
            (
                "<http://ex/e>",
                "<http://ex/label>",
                "\"tab\\there\\nline two\"^^<http://www.w3.org/2001/XMLSchema#string>",
            ),
        ]);
        let rete = Rete::open(&bytes).unwrap();
        let subjects = |filter: &str| {
            let q = format!("SELECT ?s WHERE {{ ?s <http://ex/label> ?l FILTER({filter}) }}");
            let (_, sols) = eval_sparql(&rete, &q).unwrap();
            let mut out: Vec<String> = sols.iter().map(|b| b["s"].clone()).collect();
            out.sort();
            out
        };
        let (a, b, c, e) = (
            "<http://ex/a>",
            "<http://ex/b>",
            "<http://ex/c>",
            "<http://ex/e>",
        );
        assert_eq!(subjects(r#"CONTAINS(?l, "Quantum")"#), vec![a, b, c]);
        assert_eq!(subjects(r#"CONTAINS(?l, "Gases")"#), vec![c]);
        assert_eq!(subjects(r#"STRENDS(?l, "loudly")"#), vec![b]);
        assert_eq!(subjects(r#"STRSTARTS(?l, "Theory of \"Q")"#), vec![c]);
        assert_eq!(subjects(r#"REGEX(?l, "gases$", "i")"#), vec![c]);
        assert_eq!(subjects(r#"REGEX(?l, "^line two$", "m")"#), vec![e]);
        assert_eq!(subjects(r#"CONTAINS(?l, "\there")"#), vec![e]);
        assert_eq!(subjects(r#"LANGMATCHES(LANG(?l), "en")"#), vec![b]);
        assert_eq!(subjects(r#"STRAFTER(?l, "Quantum") = "\" Gases""#), vec![c]);
        assert_eq!(
            subjects(r#"STRBEFORE(?l, " loudly") = "He said \"Quantum\""@en"#),
            vec![b]
        );
        assert_eq!(subjects(r#"STRLEN(?l) = 25"#), vec![c]);
        assert_eq!(subjects(r#"STRLEN(?l) = 24"#), vec![b]);
    }

    /// The OWL 2 QL rewrite is gated PER ATOM: one `rdfs:domain` axiom on a
    /// property the data never uses used to add a domain branch to every class
    /// atom of every reasoned query (0.30 s became 1.02 s, 4.2 MB became
    /// 11.8 MB fetched, on OpenCitations with its ontology). A class with no
    /// subclass and no domain/range property the file uses keeps its plain
    /// pattern; a class that gains members becomes a `DistinctUnion` holding
    /// exactly the sources that can produce them.
    #[test]
    fn reasoning_rewrites_only_the_atoms_it_can_extend() {
        let dom = "<http://www.w3.org/2000/01/rdf-schema#domain>";
        let sub = "<http://www.w3.org/2000/01/rdf-schema#subClassOf>";
        let ty = "<http://www.w3.org/1999/02/22-rdf-syntax-ns#type>";
        let bytes = rete_from(&[
            (
                "<http://ex/hasCharacterisation>",
                dom,
                "<http://ex/Citation>",
            ),
            ("<http://ex/SelfCitation>", sub, "<http://ex/Citation>"),
            ("<http://ex/writtenBy>", dom, "<http://ex/Work>"),
            ("<http://ex/r1>", ty, "<http://ex/Role>"),
            ("<http://ex/w1>", ty, "<http://ex/Work>"),
            ("<http://ex/w2>", "<http://ex/writtenBy>", "<http://ex/ann>"),
        ]);
        let rete = Rete::open(&bytes).unwrap();
        let rewrite = |q: &str| {
            let sel = parse_select(q).unwrap();
            ql::reason_rewrite(sel.plan, &rete, &sel.project)
        };
        // No subclass, and the one domain axiom is about another class: as is.
        assert!(matches!(
            rewrite("SELECT ?s WHERE { ?s a <http://ex/Role> ; ?p ?o }"),
            Plan::Bgp(p) if p.len() == 2
        ));
        // The domain axiom's own class: its property and its subclass are
        // never used, so there is nothing to add.
        assert!(matches!(
            rewrite("SELECT ?s WHERE { ?s a <http://ex/Citation> }"),
            Plan::Bgp(p) if p.len() == 1
        ));
        // Work gains the subjects of writtenBy: two sources, one per way.
        match rewrite("SELECT ?s WHERE { ?s a <http://ex/Work> }") {
            Plan::DistinctUnion { vars, sources } => {
                assert_eq!(vars, vec!["s".to_string()]);
                assert_eq!(sources.len(), 2);
            }
            other => panic!("expected a DistinctUnion, got {other:?}"),
        }
        let (_, sols) =
            eval_sparql_reasoned(&rete, "SELECT ?s WHERE { ?s a <http://ex/Work> }").unwrap();
        let mut got: Vec<&str> = sols.iter().map(|b| b["s"].as_str()).collect();
        got.sort();
        assert_eq!(got, vec!["<http://ex/w1>", "<http://ex/w2>"]);
    }

    /// A join whose left side turns out far smaller than the right BGP probes
    /// the BGP per left row instead of evaluating it whole — the same rows
    /// either way. Driven here with the probe floor removed, so the small
    /// graph takes the path a 2.9 B-triple file takes for real.
    #[test]
    fn a_small_left_side_probes_the_right_bgp() {
        let mut t: Vec<(String, String, String)> = Vec::new();
        let e = |n: String| format!("<http://ex/{n}>");
        for i in 0..200 {
            t.push((e(format!("j{i}")), e("a".into()), e("Journal".into())));
            t.push((e(format!("j{i}")), e("title".into()), format!("\"T{i}\"")));
            t.push((e(format!("j{i}")), e("id".into()), e(format!("id{i}"))));
            t.push((e(format!("id{i}")), e("value".into()), format!("\"{i}\"")));
        }
        t.push((e("work".into()), e("partOf".into()), e("issue".into())));
        t.push((e("issue".into()), e("partOf".into()), e("j7".into())));
        let refs: Vec<(&str, &str, &str)> = t
            .iter()
            .map(|(s, p, o)| (s.as_str(), p.as_str(), o.as_str()))
            .collect();
        let rete = Rete::open(&rete_from(&refs)).unwrap();
        let queries = [
            "PREFIX ex: <http://ex/> SELECT ?j ?t ?v WHERE { ex:work ex:partOf+ ?j . \
             ?j ex:a ex:Journal ; ex:title ?t ; ex:id ?i . ?i ex:value ?v }",
            "PREFIX ex: <http://ex/> SELECT ?j ?t WHERE { ex:work ex:partOf+ ?j \
             OPTIONAL { ?j ex:title ?t } }",
        ];
        for q in queries {
            let run = |floor: f64| {
                crate::bgp::with_probe_floor(floor, || {
                    let (_, sols) = eval_sparql(&rete, q).unwrap();
                    let mut v: Vec<String> = sols.iter().map(|b| format!("{b:?}")).collect();
                    v.sort();
                    v
                })
            };
            let hashed = run(f64::MAX);
            assert!(!hashed.is_empty());
            assert_eq!(run(0.0), hashed, "{q}");
        }
    }
    /// 200 people: each `knows` the next, the even ones have an `age`, every
    /// tenth is a `Manager`; two of them carry a second type.
    fn people() -> Rete {
        const TYPE: &str = "<http://www.w3.org/1999/02/22-rdf-syntax-ns#type>";
        let e = |n: String| format!("<http://ex/{n}>");
        let mut t: Vec<(String, String, String)> = Vec::new();
        for i in 0..200 {
            t.push((
                e(format!("p{i}")),
                e("knows".into()),
                e(format!("p{}", (i + 1) % 200)),
            ));
            t.push((e(format!("p{i}")), e("name".into()), format!("\"N{i}\"")));
            if i % 2 == 0 {
                t.push((e(format!("p{i}")), e("age".into()), format!("\"{i}\"")));
            }
            let class = if i % 10 == 0 { "Manager" } else { "Person" };
            t.push((e(format!("p{i}")), TYPE.into(), e(class.into())));
        }
        for i in [3, 4] {
            t.push((e(format!("p{i}")), TYPE.into(), e("Agent".into())));
        }
        let refs: Vec<(&str, &str, &str)> = t
            .iter()
            .map(|(s, p, o)| (s.as_str(), p.as_str(), o.as_str()))
            .collect();
        Rete::open(&rete_from(&refs)).unwrap()
    }

    /// Rows of `q` as sorted strings, under the given probe floor
    /// (`f64::MAX`: the hash and whole-evaluation plans; `0.0`: every probe
    /// path a 2.9 B-triple file takes for real).
    fn rows_at(rete: &Rete, q: &str, floor: f64) -> Vec<String> {
        crate::bgp::with_probe_floor(floor, || {
            let (_, sols) = eval_sparql(rete, q).unwrap();
            let mut v: Vec<String> = sols.iter().map(|b| format!("{b:?}")).collect();
            v.sort();
            v
        })
    }

    /// `FILTER (NOT) EXISTS` and `MINUS` over a fat BGP probe it per outer
    /// row: the same rows as loading it, in every shape.
    #[test]
    fn correlated_exists_and_minus_match_whole_evaluation() {
        let rete = people();
        let pre = "PREFIX ex: <http://ex/> ";
        let bodies = [
            // NOT EXISTS / EXISTS on a shared subject
            "SELECT ?s WHERE { ?s ex:knows ?o . FILTER NOT EXISTS { ?s ex:age ?a } }",
            "SELECT ?s WHERE { ?s ex:knows ?o . FILTER EXISTS { ?s ex:age ?a } }",
            // two inner patterns, joined through a variable the outer side lacks
            "SELECT ?s WHERE { ?s ex:name ?n . FILTER NOT EXISTS { ?s ex:knows ?o . ?o ex:age ?a } }",
            // shared object, an unrelated inner variable
            "SELECT ?o WHERE { ?s ex:knows ?o . FILTER NOT EXISTS { ?x ex:knows ?o . ?x ex:age ?a } }",
            // an inner constant the file does not hold: EXISTS is false for every row
            "SELECT ?s WHERE { ?s ex:name ?n . FILTER NOT EXISTS { ?s ex:nothere ?a } }",
            "SELECT ?s WHERE { ?s ex:name ?n . FILTER EXISTS { ?s ex:nothere ?a } }",
            // no variable shared: the whole-evaluation answer
            "SELECT ?s WHERE { ?s ex:name ?n . FILTER NOT EXISTS { ?z ex:age \"4\" } }",
            // a ground inner pattern
            "SELECT ?s WHERE { ?s ex:name ?n . FILTER EXISTS { ex:p4 ex:age \"4\" } }",
            // an outer value the inner side can never hold
            "SELECT ?s WHERE { ?s ex:name ?n . BIND(CONCAT(?n, \"!\") AS ?m) FILTER NOT EXISTS { ?m ex:age ?a } }",
            // MINUS: shared subject, shared object, disjoint domain, unknown constant
            "SELECT ?s WHERE { ?s ex:knows ?o . MINUS { ?s ex:age ?a } }",
            "SELECT ?o WHERE { ?s ex:knows ?o . MINUS { ?x ex:knows ?o . ?x ex:age ?a } }",
            "SELECT ?s WHERE { ?s ex:name ?n . MINUS { ?z ex:age ?a } }",
            "SELECT ?s WHERE { ?s ex:name ?n . MINUS { ?s ex:nothere ?a } }",
            // OPTIONAL on the left: MINUS keeps the rows that share no binding
            "SELECT ?s ?a WHERE { ?s ex:name ?n . OPTIONAL { ?s ex:age ?a } MINUS { ?a ex:nothing ?z } }",
            // under VALUES
            "SELECT ?s WHERE { VALUES ?s { ex:p1 ex:p2 ex:p3 } FILTER NOT EXISTS { ?s ex:age ?a } }",
        ];
        for body in bodies {
            let q = format!("{pre}{body}");
            let whole = rows_at(&rete, &q, f64::MAX);
            assert_eq!(rows_at(&rete, &q, 0.0), whole, "{body}");
        }
        // The answers themselves, so "same as before" is not "both empty".
        let n = |body: &str| rows_at(&rete, &format!("{pre}{body}"), 0.0).len();
        assert_eq!(n(bodies[0]), 100);
        assert_eq!(n(bodies[9]), 100);
        assert_eq!(n(bodies[11]), 200);
    }

    /// A UNION joined with a small bound side takes that side's bindings into
    /// each branch (a `VALUES`, or a preceding pattern): the same rows as
    /// evaluating every branch whole.
    #[test]
    fn a_union_receives_the_bindings_of_the_side_joined_with_it() {
        let rete = people();
        let pre = "PREFIX ex: <http://ex/> ";
        let bodies = [
            "SELECT ?r ?p ?v WHERE { VALUES ?r { ex:p1 ex:p2 ex:p3 ex:nobody } \
             { ?r ?p ?v } UNION { ?r ex:knows ?n . ?n ?p ?v } }",
            "SELECT ?r ?p ?v WHERE { { ?r ?p ?v } UNION { ?r ex:knows ?n . ?n ?p ?v } \
             VALUES ?r { ex:p1 ex:p2 ex:p3 } }",
            // three branches, one with an unknown constant
            "SELECT ?r ?v WHERE { VALUES ?r { ex:p4 ex:p8 } { ?r ex:age ?v } UNION \
             { ?r ex:nothere ?v } UNION { ?r ex:knows ?v } }",
            // a preceding pattern is the bound side
            "SELECT ?x ?v WHERE { ?x ex:age \"4\" . { ?x ex:knows ?v } UNION { ?x ex:name ?v } }",
            // a variable only some branches bind
            "SELECT ?r ?v ?w WHERE { VALUES ?r { ex:p3 ex:p6 } { ?r ex:knows ?v } UNION \
             { ?r ex:age ?w } }",
            // UNDEF in the VALUES block: no certain binding, the plain join
            "SELECT ?r ?v WHERE { VALUES ?r { ex:p3 UNDEF } { ?r ex:age ?v } UNION { ?r ex:name ?v } }",
        ];
        for body in bodies {
            let q = format!("{pre}{body}");
            let whole = rows_at(&rete, &q, f64::MAX);
            assert!(!whole.is_empty(), "{body}");
            assert_eq!(rows_at(&rete, &q, 0.0), whole, "{body}");
        }
        // p1: knows, name, type; p2: knows, name, age, type; p3: those and a
        // second type = 11 rows from the first branch; the second branch walks
        // to p2 (4), p3 (4) and p4 (5: age, a second type).
        let q = format!("{pre}{}", bodies[0]);
        assert_eq!(rows_at(&rete, &q, 0.0).len(), 11 + 13);
    }

    /// An unbound class pattern whose subject an earlier pattern binds is
    /// probed per subject; the same rows as scanning every type statement.
    #[test]
    fn an_unbound_class_joined_on_a_bound_subject_probes_per_subject() {
        let rete = people();
        let pre = "PREFIX ex: <http://ex/> ";
        let bodies = [
            "SELECT ?s ?c WHERE { ?s ex:age ?a . ?s a ?c }",
            "SELECT ?s ?o ?c WHERE { ?s ex:knows ?o . ?o a ?c }",
            "SELECT ?c (COUNT(*) AS ?n) WHERE { ?s ex:age ?a . ?s a ?c } GROUP BY ?c",
            // a class variable bound on both sides
            "SELECT ?s ?c WHERE { ?s ex:knows ?o . ?o a ?c . ?s a ?c }",
            // nothing binds the subject: the scan stays
            "SELECT ?s ?c WHERE { ?s ?p ?x . ?s a ?c . FILTER(?p = ex:age) }",
        ];
        for body in bodies {
            let q = format!("{pre}{body}");
            let whole = rows_at(&rete, &q, f64::MAX);
            assert!(!whole.is_empty(), "{body}");
            assert_eq!(rows_at(&rete, &q, 0.0), whole, "{body}");
        }
        // 100 even people; p4 has a second type.
        let q = format!("{pre}{}", bodies[0]);
        assert_eq!(rows_at(&rete, &q, 0.0).len(), 101);
    }

    /// A pattern inside `GRAPH <g>` is sized by the graph, not by the file.
    #[test]
    fn a_named_graph_is_sized_by_its_own_triples() {
        use crate::write_dataset_v05 as write_dataset;
        let mut db = DictionaryBuilder::new();
        let big: Vec<(String, String, String)> = (0..300)
            .map(|i| {
                (
                    format!("<http://ex/s{i}>"),
                    "<http://ex/p>".to_string(),
                    format!("<http://ex/o{i}>"),
                )
            })
            .collect();
        let tiny = (
            "<http://ex/s0>".to_string(),
            "<http://ex/q>".to_string(),
            "<http://ex/o0>".to_string(),
        );
        for (s, p, o) in big.iter().chain(std::iter::once(&tiny)) {
            db.observe(s, p, o);
        }
        let dict = db.build();
        let mut default = GraphIndexBuilder::new();
        for (s, p, o) in &big {
            default.push(dict.encode(s, p, o).unwrap());
        }
        let mut small = GraphIndexBuilder::new();
        small.push(dict.encode(&tiny.0, &tiny.1, &tiny.2).unwrap());
        let named = vec![("<http://ex/g>".to_string(), small.build())];
        let bytes = write_dataset(&dict, &default.build(), &named, true, &[], 0);
        let rete = Rete::open(&bytes).unwrap();
        let g = rete.graph_index("<http://ex/g>").unwrap();
        assert_eq!(rete.index_triples(g), 1, "the graph's own count");
        assert_eq!(
            rete.index_triples(rete.default_index()),
            rete.header().quad_count,
            "the default graph keeps the file's total"
        );
        let (_, sols) = eval_sparql(
            &rete,
            "SELECT ?o WHERE { GRAPH <http://ex/g> { <http://ex/s0> <http://ex/q> ?o } }",
        )
        .unwrap();
        assert_eq!(sols.len(), 1);
    }

    /// The same, in the quad index: a graph's view takes its quad count from
    /// the directory.
    #[test]
    fn a_quad_index_graph_is_sized_by_its_directory_count() {
        use crate::ingest::{assemble_dataset_with_layout, GraphLayout};
        let q = |s: &str, g: Option<&str>| -> crate::ingest::RawQuad {
            (
                format!("<http://ex/{s}>"),
                "<http://ex/p>".to_string(),
                "<http://ex/o>".to_string(),
                g.map(|g| format!("<http://ex/{g}>")),
            )
        };
        let mut quads: Vec<_> = (0..50).map(|i| q(&format!("d{i}"), None)).collect();
        quads.extend((0..7).map(|i| q(&format!("a{i}"), Some("ga"))));
        quads.extend((0..3).map(|i| q(&format!("b{i}"), Some("gb"))));
        let (bytes, _) = assemble_dataset_with_layout(
            quads,
            false,
            false,
            None,
            crate::pyramid::PyramidAlgo::Louvain,
            crate::index::PermSet::ALL,
            GraphLayout::Quad,
            |_, _| Vec::<u8>::new(),
        );
        let rete = Rete::open(&bytes).unwrap();
        assert_eq!(
            rete.index_triples(rete.graph_index("<http://ex/ga>").unwrap()),
            7
        );
        assert_eq!(
            rete.index_triples(rete.graph_index("<http://ex/gb>").unwrap()),
            3
        );
        let (_, sols) = eval_sparql(
            &rete,
            "SELECT ?s WHERE { GRAPH <http://ex/ga> { ?s <http://ex/p> <http://ex/o> . \
             FILTER NOT EXISTS { ?s <http://ex/p> <http://ex/zz> } } }",
        )
        .unwrap();
        assert_eq!(sols.len(), 7);
    }
}