oxilite-core 0.9.0

//! SPARQL algebra → SQL.
//!
//! Each graph pattern compiles to a [`Block`]: FROM items, WHERE conditions, variable
//! bindings and solution-modifier state. Blocks stay "plain" (a single SELECT-FROM-WHERE)
//! for as long as possible so that SQLite sees one flat join; they are sealed into
//! subqueries only when SQL semantics require it.
//!
// @lat: [[architecture#SPARQL to SQL compiler]]

pub mod expr;
mod ops;
pub mod plan;
mod vector;

use crate::encoding::{encode_literal, named_node_id, term_id, EncodedRows, DEFAULT_GRAPH_ID};
use crate::error::{Error, Result};
use crate::reason::{Entailment, GraphFilter};
use crate::sql::Capabilities;
use crate::stats::Stats;
use expr::V;
use oxrdf::{Literal, Term, Variable};
use plan::Pos;
use spargebra::algebra::{Expression, GraphPattern, OrderExpression, QueryDataset};
use spargebra::term::{GroundTerm, NamedNodePattern, TermPattern, TriplePattern};
use std::collections::{BTreeMap, HashMap, HashSet};
use std::fmt::Write;

/// Per-query options.
#[derive(Debug, Clone, PartialEq, Eq)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
#[cfg_attr(feature = "serde", serde(default, rename_all = "camelCase"))]
pub struct QueryOptions {
    /// Treat the default graph as the union of all graphs (Oxigraph's `union_default_graph`).
    pub union_default_graph: bool,
    /// Let SQLite choose the join order instead of the oxilite planner (for benchmarking).
    pub sqlite_planner: bool,
    /// Overrides the default graph with the merge of these graphs (term ids, 0 = default
    /// graph), like Oxigraph's `default_graph` option.
    pub default_graph: Option<Vec<i64>>,
    /// Overrides the available named graphs (term ids).
    pub named_graphs: Option<Vec<i64>>,
    /// Entailment regime (default: none, like Oxigraph).
    pub reasoning: crate::reason::Reasoning,
    /// Also match materialized inferences (`quads_inf`, see `materialize()`).
    pub include_inferred: bool,
    /// Match triples of the registered schema graphs (ontologies, shapes; see
    /// `oxilite_core::registry`). Default `true`: the dataset is the dataset. Set to `false`
    /// to query the data without the axioms and shapes that describe it.
    pub include_schema_graphs: bool,
    /// Value types known for variables (e.g. from a schema): comparisons on them compile to
    /// one typed comparison instead of a comparison per possible type. A value of another
    /// type then compares as unknown (false in a filter).
    pub var_types: BTreeMap<String, ValueType>,
    /// Read the store as it was at this version (`HEAD~2`, `#42`, `@2026-09-01T00:00:00Z`; see
    /// `version::VersionRef`). Needs a store with a change log; the store resolves it to
    /// [`Self::as_of_tick`] before compiling.
    pub as_of: Option<String>,
    /// The resolved tick of [`Self::as_of`].
    pub as_of_tick: Option<i64>,
    /// Resolved ticks of the versions named by `SERVICE <oxilite:version/REF>` (by `REF`).
    pub versions: BTreeMap<String, i64>,
    /// Host functions callable by IRI (see [`crate::functions`]). A call to one is evaluated
    /// by the fallback evaluator; the rest of the query still compiles to SQL.
    #[cfg_attr(feature = "serde", serde(skip))]
    pub functions: crate::functions::Functions,
}

/// The IRI prefix naming a version of the store in `SERVICE <oxilite:version/REF> { … }`.
pub const VERSION_SERVICE: &str = "oxilite:version/";

/// The id offset of inline integers (a tick `t` is the integer id `t + expr_int_base()`).
pub fn expr_int_base() -> i64 {
    expr::INT_BASE
}

/// The graph of the store's history: commits (PROV-O activities, identified by their tick as an
/// `xsd:integer`) and their changes (`oxl:added` / `oxl:removed` triple terms).
pub const HISTORY_GRAPH: &str = "oxilite:history";

impl Default for QueryOptions {
    fn default() -> Self {
        Self {
            union_default_graph: false,
            sqlite_planner: false,
            default_graph: None,
            named_graphs: None,
            reasoning: crate::reason::Reasoning::default(),
            include_inferred: false,
            // The dataset is the dataset: schema graphs are matched unless asked otherwise.
            include_schema_graphs: true,
            var_types: BTreeMap::new(),
            as_of: None,
            as_of_tick: None,
            versions: BTreeMap::new(),
            functions: crate::functions::Functions::default(),
        }
    }
}

/// A static value type for [`QueryOptions::var_types`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
#[cfg_attr(feature = "serde", serde(rename_all = "camelCase"))]
pub enum ValueType {
    /// Numbers (`xsd:integer`, `xsd:decimal`, `xsd:double`… and derived types).
    Numeric,
    /// Simple literals (`xsd:string`).
    String,
    Boolean,
}

/// How a variable is represented in SQL.
#[derive(Debug, Clone)]
pub(crate) enum Col {
    /// A term id column.
    Id(String),
    /// A computed value (see [`V`]).
    Val(Box<V>),
}

impl Col {
    /// SQL of the term id usable as a join key.
    pub(crate) fn key(&self) -> Option<&str> {
        match self {
            Self::Id(x) => Some(x),
            Self::Val(v) => v.id.as_deref(),
        }
    }

    pub(crate) fn value(&self) -> V {
        match self {
            Self::Id(x) => V::from_id(x),
            Self::Val(v) => (**v).clone(),
        }
    }
}

#[derive(Debug, Clone)]
pub(crate) struct Binding {
    pub col: Col,
    pub nullable: bool,
    /// The column is an expression (not a plain column of a FROM item); such bindings must
    /// be sealed before being placed on the right of a LEFT JOIN.
    #[allow(dead_code)]
    pub computed: bool,
    /// Already equated with an outer (EXISTS) binding.
    pub correlated: bool,
}

impl Binding {
    fn id(sql: impl Into<String>) -> Self {
        Self {
            col: Col::Id(sql.into()),
            nullable: false,
            computed: false,
            correlated: false,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum Join {
    First,
    Cross,
    Inner,
    /// `LEFT JOIN … ON (condition)` (OPTIONAL).
    #[allow(dead_code)]
    Left(String),
}

#[derive(Debug, Clone)]
pub(crate) struct FromItem {
    pub join: Join,
    pub item: String,
}

#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, PartialOrd, Ord)]
pub(crate) enum Stage {
    #[default]
    Plain,
    Grouped,
    Ordered,
    Distinct,
    Sliced,
}

#[derive(Debug, Clone, Default)]
pub(crate) struct Block {
    pub from: Vec<FromItem>,
    pub wheres: Vec<String>,
    pub cols: BTreeMap<usize, Binding>,
    pub group_by: Option<Vec<String>>,
    pub order_by: Vec<String>,
    pub distinct: bool,
    pub limit: Option<usize>,
    pub offset: usize,
    pub stage: Stage,
    /// Extra select-list entries (e.g. the aggregate that makes a bare column meaningful).
    pub extra_select: Vec<String>,
    /// Render `LIMIT -1` so that SQLite does not flatten this subquery into an aggregate
    /// query (flattening restriction 21): its computed columns are then evaluated once per row
    /// instead of being inlined into every aggregate that reads them.
    pub no_flatten: bool,
}

pub(crate) const VAL_FIELDS: [&str; 10] = ["i", "k", "l", "d", "g", "n", "t", "s", "b", "x"];

pub(crate) fn val_fields(v: &V) -> [String; 10] {
    [
        v.id.clone().unwrap_or_else(|| "NULL".into()),
        v.kind.clone(),
        if v.computed_num {
            "NULL".into()
        } else {
            v.lex.clone()
        },
        v.dt.clone(),
        v.lang.clone(),
        v.num.clone(),
        v.nt.clone(),
        v.ts.clone(),
        v.boolv.clone(),
        v.aux.clone(),
    ]
}

impl Block {
    pub(crate) fn is_plain(&self) -> bool {
        self.stage == Stage::Plain
    }

    pub(crate) fn is_unit(&self) -> bool {
        self.from.is_empty() && self.wheres.is_empty() && self.cols.is_empty()
    }

    pub(crate) fn render_from(items: &[FromItem]) -> String {
        let mut out = String::new();
        for (i, f) in items.iter().enumerate() {
            if i == 0 {
                out.push_str(&f.item);
                continue;
            }
            match &f.join {
                Join::First | Join::Inner => {
                    let _ = write!(out, " JOIN {}", f.item);
                }
                Join::Cross => {
                    let _ = write!(out, " CROSS JOIN {}", f.item);
                }
                Join::Left(on) => {
                    let _ = write!(out, " LEFT JOIN {} ON ({on})", f.item);
                }
            }
        }
        out
    }

    /// Renders the select list entries of the given variables.
    fn select_list(&self, vars: &[usize], text_ids: bool) -> Vec<String> {
        let mut sel = Vec::new();
        for idx in vars {
            match self.cols.get(idx).map(|b| &b.col) {
                Some(Col::Id(x)) => sel.push(if text_ids {
                    format!("CAST({x} AS TEXT) AS v{idx}")
                } else {
                    format!("{x} AS v{idx}")
                }),
                Some(Col::Val(v)) => {
                    for (suffix, field) in VAL_FIELDS.iter().zip(val_fields(v)) {
                        if text_ids && *suffix == "i" {
                            sel.push(format!("CAST({field} AS TEXT) AS v{idx}_{suffix}"));
                        } else {
                            sel.push(format!("{field} AS v{idx}_{suffix}"));
                        }
                    }
                }
                None => sel.push(format!("NULL AS v{idx}")),
            }
        }
        sel
    }

    /// Renders everything after the select list: FROM, WHERE, GROUP BY, ORDER BY, LIMIT.
    pub(crate) fn tail(&self) -> String {
        let mut sql = String::new();
        if !self.from.is_empty() {
            sql.push_str(" FROM ");
            sql.push_str(&Self::render_from(&self.from));
        }
        if !self.wheres.is_empty() {
            sql.push_str(" WHERE ");
            sql.push_str(&self.wheres.join(" AND "));
        }
        if let Some(g) = &self.group_by {
            if !g.is_empty() {
                sql.push_str(" GROUP BY ");
                sql.push_str(&g.join(", "));
            }
        }
        if !self.order_by.is_empty() {
            sql.push_str(" ORDER BY ");
            sql.push_str(&self.order_by.join(", "));
        }
        if self.limit.is_some() || self.offset > 0 || self.no_flatten {
            let _ = write!(
                sql,
                " LIMIT {}",
                self.limit.map_or_else(|| "-1".into(), |l| l.to_string())
            );
            if self.offset > 0 {
                let _ = write!(sql, " OFFSET {}", self.offset);
            }
        }
        sql
    }

    /// Renders the block as a SELECT statement.
    pub(crate) fn to_select(&self, vars: Option<&[usize]>, text_ids: bool) -> String {
        let all: Vec<usize> = self.cols.keys().copied().collect();
        let mut sel = self.select_list(vars.unwrap_or(&all), text_ids);
        sel.extend(self.extra_select.iter().cloned());
        if sel.is_empty() {
            sel.push("1 AS _u".into());
        }
        format!(
            "SELECT {}{}{}",
            if self.distinct { "DISTINCT " } else { "" },
            sel.join(", "),
            self.tail()
        )
    }
}

#[derive(Debug, Clone)]
enum DefaultGraph {
    Zero,
    Union,
    List(Vec<i64>),
}

#[derive(Debug, Clone)]
struct Dataset {
    default: DefaultGraph,
    named: Option<Vec<i64>>,
}

#[derive(Debug, Clone, Copy)]
enum GraphScope {
    Default,
    Fixed(i64),
    Var(usize),
    /// `GRAPH <oxilite:history>`: the store's commits and changes (see `version::history_sql`).
    History,
}

/// The SPARQL → SQL compiler state for one query or update.
pub(crate) struct Compiler<'a> {
    pub stats: &'a Stats,
    pub caps: &'a Capabilities,
    pub options: &'a QueryOptions,
    vars: HashMap<Variable, usize>,
    pub var_names: Vec<Variable>,
    aliases: usize,
    /// Constants appearing in the query: decodable without a lookup.
    pub constants: HashMap<i64, Term>,
    /// Bindings visible from enclosing EXISTS scopes.
    pub outer: Vec<BTreeMap<usize, Binding>>,
    pub now: Literal,
    pub base_iri: Option<String>,
    dataset: Dataset,
    scope: GraphScope,
    /// Rows needed by constants that end up stored (update templates).
    pub rows: EncodedRows,
    /// Planner decisions and warnings, for `explain()`.
    pub notes: Vec<String>,
    /// Triple-term patterns awaiting a `triple_terms` join (variable, pattern).
    pending_triples: Vec<(usize, TriplePattern)>,
    /// Variables bound by an enclosing join (e.g. the left side of OPTIONAL): the planner
    /// orders patterns as if they were bound, since SQLite evaluates the right side per row.
    pub(crate) plan_hint: Vec<HashSet<usize>>,
    /// The tick the quads are read at (`as_of`, or the version of an enclosing
    /// `SERVICE <oxilite:version/…>`); `None` reads the current store.
    pub(crate) as_of: Option<i64>,
}

impl<'a> Compiler<'a> {
    pub(crate) fn new(
        stats: &'a Stats,
        caps: &'a Capabilities,
        options: &'a QueryOptions,
        dataset: Option<&QueryDataset>,
        base_iri: Option<String>,
    ) -> Self {
        let mut dataset = match dataset {
            Some(ds) => Dataset {
                default: DefaultGraph::List(
                    ds.default
                        .iter()
                        .map(|g| named_node_id(g.as_str()))
                        .collect(),
                ),
                // `None` (e.g. `WITH`) leaves every named graph available, as in spareval.
                named: ds
                    .named
                    .as_ref()
                    .map(|n| n.iter().map(|g| named_node_id(g.as_str())).collect()),
            },
            None => Dataset {
                default: if options.union_default_graph {
                    DefaultGraph::Union
                } else {
                    DefaultGraph::Zero
                },
                named: None,
            },
        };
        if let Some(d) = &options.default_graph {
            dataset.default = DefaultGraph::List(d.clone());
        }
        if let Some(n) = &options.named_graphs {
            dataset.named = Some(n.clone());
        }
        Self {
            stats,
            caps,
            options,
            vars: HashMap::new(),
            var_names: Vec::new(),
            aliases: 0,
            constants: HashMap::new(),
            outer: Vec::new(),
            now: Literal::from(oxsdatatypes::DateTime::now()),
            base_iri,
            dataset,
            scope: GraphScope::Default,
            rows: EncodedRows::default(),
            notes: Vec::new(),
            pending_triples: Vec::new(),
            plan_hint: Vec::new(),
            as_of: options.as_of_tick,
        }
    }

    /// Joins `terms` once per stored-term variable an expression reads, so that its fields
    /// come from one row instead of one correlated lookup each (see [`V::from_joined`]).
    /// Returns the bindings to restore once the expression is compiled: other operators keep
    /// seeing plain id columns.
    pub(crate) fn join_terms(
        &mut self,
        b: &mut Block,
        e: &Expression,
    ) -> Vec<(usize, Binding, String)> {
        if matches!(e, Expression::Variable(_) | Expression::Bound(_)) {
            return Vec::new();
        }
        let mut vars = Vec::new();
        expression_variables(e, &mut vars);
        self.join_term_vars(b, &vars)
    }

    /// [`Self::join_terms`] for a list of variables.
    pub(crate) fn join_term_vars(
        &mut self,
        b: &mut Block,
        vars: &[Variable],
    ) -> Vec<(usize, Binding, String)> {
        let mut restore = Vec::new();
        if b.stage >= Stage::Grouped || b.from.is_empty() || !b.is_plain() {
            return restore;
        }
        for v in vars {
            let Some(&idx) = self.vars.get(v) else {
                continue;
            };
            let Some(bind) = b.cols.get(&idx) else {
                continue;
            };
            let Col::Id(x) = &bind.col else {
                continue;
            };
            if bind.correlated || bind.computed {
                continue;
            }
            let x = x.clone();
            let t = self.alias("t");
            b.from.push(FromItem {
                join: Join::Left(format!("{t}.id = {x}")),
                item: format!("terms {t}"),
            });
            let mut joined = bind.clone();
            let v = V::from_joined(&x, &t);
            let marker = v.lex.clone();
            joined.col = Col::Val(Box::new(v));
            restore.push((idx, bind.clone(), marker));
            b.cols.insert(idx, joined);
        }
        restore
    }

    /// Puts back the id bindings replaced by [`Self::join_terms`], unless the block was sealed
    /// meanwhile (its bindings then name the subquery's columns).
    pub(crate) fn restore_terms(b: &mut Block, restore: Vec<(usize, Binding, String)>) {
        for (i, bind, marker) in restore {
            if let Some(cur) = b.cols.get_mut(&i) {
                if matches!(&cur.col, Col::Val(v) if v.lex == marker) {
                    *cur = bind;
                }
            }
        }
    }

    /// Keeps SQL shallow for nested function calls: a large argument (e.g. the canonical string
    /// of a decimal inside `xsd:float(xsd:string(?price))`) is computed once as a column of a
    /// sealed subquery, instead of being inlined everywhere the outer function uses it, which
    /// multiplies SQL size (and per-row work) at each nesting level.
    pub(crate) fn shallow(&mut self, b: Block, e: &Expression) -> Result<(Block, Expression)> {
        use Expression as E;
        let two = |me: &mut Self, b: Block, x: &E, y: &E| -> Result<(Block, E, E)> {
            let (b, x) = me.shallow_arg(b, x)?;
            let (b, y) = me.shallow_arg(b, y)?;
            Ok((b, x, y))
        };
        Ok(match e {
            E::FunctionCall(f, args) => {
                let mut b = b;
                let mut out = Vec::with_capacity(args.len());
                for a in args {
                    let (nb, a) = self.shallow_arg(b, a)?;
                    b = nb;
                    out.push(a);
                }
                (b, E::FunctionCall(f.clone(), out))
            }
            E::Equal(x, y) => {
                let (b, x, y) = two(self, b, x, y)?;
                (b, E::Equal(Box::new(x), Box::new(y)))
            }
            E::Less(x, y) => {
                let (b, x, y) = two(self, b, x, y)?;
                (b, E::Less(Box::new(x), Box::new(y)))
            }
            E::LessOrEqual(x, y) => {
                let (b, x, y) = two(self, b, x, y)?;
                (b, E::LessOrEqual(Box::new(x), Box::new(y)))
            }
            E::Greater(x, y) => {
                let (b, x, y) = two(self, b, x, y)?;
                (b, E::Greater(Box::new(x), Box::new(y)))
            }
            E::GreaterOrEqual(x, y) => {
                let (b, x, y) = two(self, b, x, y)?;
                (b, E::GreaterOrEqual(Box::new(x), Box::new(y)))
            }
            E::Add(x, y) => {
                let (b, x, y) = two(self, b, x, y)?;
                (b, E::Add(Box::new(x), Box::new(y)))
            }
            E::Subtract(x, y) => {
                let (b, x, y) = two(self, b, x, y)?;
                (b, E::Subtract(Box::new(x), Box::new(y)))
            }
            E::Multiply(x, y) => {
                let (b, x, y) = two(self, b, x, y)?;
                (b, E::Multiply(Box::new(x), Box::new(y)))
            }
            E::Divide(x, y) => {
                let (b, x, y) = two(self, b, x, y)?;
                (b, E::Divide(Box::new(x), Box::new(y)))
            }
            E::And(x, y) => {
                let (b, x) = self.shallow(b, x)?;
                let (b, y) = self.shallow(b, y)?;
                (b, E::And(Box::new(x), Box::new(y)))
            }
            E::Or(x, y) => {
                let (b, x) = self.shallow(b, x)?;
                let (b, y) = self.shallow(b, y)?;
                (b, E::Or(Box::new(x), Box::new(y)))
            }
            E::Not(x) => {
                let (b, x) = self.shallow(b, x)?;
                (b, E::Not(Box::new(x)))
            }
            _ => (b, e.clone()),
        })
    }

    /// [`Self::shallow`] for an operand: after lifting inside it, a function call whose SQL is
    /// large becomes a column of a sealed subquery.
    fn shallow_arg(&mut self, b: Block, a: &Expression) -> Result<(Block, Expression)> {
        const LIMIT: usize = 1_500;
        let (mut b, a) = self.shallow(b, a)?;
        if !matches!(a, Expression::FunctionCall(..)) {
            return Ok((b, a));
        }
        let v = self.expr_term(&a, &b.cols)?;
        if v.size() <= LIMIT {
            return Ok((b, a));
        }
        let hidden = self.fresh_var("arg");
        b.cols.insert(
            hidden,
            Binding {
                col: Col::Val(Box::new(v)),
                nullable: true,
                computed: true,
                correlated: false,
            },
        );
        b = self.seal(b);
        Ok((b, Expression::Variable(self.var_names[hidden].clone())))
    }

    pub(crate) fn var(&mut self, v: &Variable) -> usize {
        if let Some(i) = self.vars.get(v) {
            return *i;
        }
        let i = self.var_names.len();
        self.vars.insert(v.clone(), i);
        self.var_names.push(v.clone());
        i
    }

    pub(crate) fn fresh_var(&mut self, hint: &str) -> usize {
        let v = Variable::new_unchecked(format!("\u{1}{hint}{}", self.var_names.len()));
        self.var(&v)
    }

    pub(crate) fn alias(&mut self, prefix: &str) -> String {
        self.aliases += 1;
        format!("{prefix}{}", self.aliases)
    }

    /// Encodes a constant term and remembers it for decoding.
    pub(crate) fn constant_id(&mut self, t: &Term) -> Result<i64> {
        let id = match t {
            Term::Triple(tr) => {
                let id = self.rows.triple(tr.as_ref().as_ref());
                self.constants.insert(id, t.clone());
                // Components are decodable too.
                self.constants.insert(
                    term_id(tr.subject.as_ref().into()),
                    tr.subject.clone().into(),
                );
                self.constants.insert(
                    named_node_id(tr.predicate.as_str()),
                    tr.predicate.clone().into(),
                );
                self.constants
                    .insert(term_id(tr.object.as_ref()), tr.object.clone());
                return Ok(id);
            }
            Term::Literal(l) => encode_literal(l.as_ref()).0,
            _ => term_id(t.as_ref()),
        };
        self.constants.insert(id, t.clone());
        Ok(id)
    }

    fn outer_binding(&self, idx: usize) -> Option<&Binding> {
        self.outer.iter().rev().find_map(|s| s.get(&idx))
    }

    /// Wraps a block into a subquery.
    pub(crate) fn seal(&mut self, b: Block) -> Block {
        let alias = self.alias("s");
        let sql = b.to_select(None, false);
        let mut cols = BTreeMap::new();
        for (idx, bind) in &b.cols {
            let col = match &bind.col {
                Col::Id(_) => Col::Id(format!("{alias}.v{idx}")),
                Col::Val(v) => Col::Val(Box::new(V {
                    id: v.id.as_ref().map(|_| format!("{alias}.v{idx}_i")),
                    kind: format!("{alias}.v{idx}_k"),
                    lex: format!("{alias}.v{idx}_l"),
                    dt: format!("{alias}.v{idx}_d"),
                    lang: format!("{alias}.v{idx}_g"),
                    num: format!("{alias}.v{idx}_n"),
                    nt: format!("{alias}.v{idx}_t"),
                    ts: format!("{alias}.v{idx}_s"),
                    boolv: format!("{alias}.v{idx}_b"),
                    stat: v.stat,
                    computed_num: false,
                    decodable: v.decodable,
                    aux: format!("{alias}.v{idx}_x"),
                    tz: "NULL".into(),
                })),
            };
            cols.insert(
                *idx,
                Binding {
                    col,
                    nullable: bind.nullable,
                    computed: false,
                    correlated: false,
                },
            );
        }
        // A sealed computed number keeps a NULL lexical form: re-derive it from num/nt.
        for (idx, bind) in &b.cols {
            if let Col::Val(v) = &bind.col {
                if v.computed_num {
                    if let Some(Binding {
                        col: Col::Val(sv), ..
                    }) = cols.get_mut(idx)
                    {
                        let n = V::numeric(sv.num.clone(), sv.nt.clone());
                        sv.lex = n.lex;
                        sv.computed_num = true;
                    }
                }
            }
        }
        Block {
            from: vec![FromItem {
                join: Join::First,
                item: format!("({sql}) AS {alias}"),
            }],
            cols,
            ..Block::default()
        }
    }

    fn plain(&mut self, b: Block) -> Block {
        if b.is_plain() {
            b
        } else {
            self.seal(b)
        }
    }

    /// Compiles a graph pattern.
    pub(crate) fn pattern(&mut self, p: &GraphPattern) -> Result<Block> {
        match p {
            GraphPattern::Bgp { patterns } => self.bgp(patterns),
            GraphPattern::Join { left, right } => {
                let a = self.pattern(left)?;
                if let GraphPattern::Path {
                    subject,
                    path,
                    object,
                } = right.as_ref()
                {
                    if let Some(b) = self.seeded_path(&a, subject, path, object)? {
                        return self.join(a, b);
                    }
                }
                let b = self.pattern(right)?;
                self.join(a, b)
            }
            GraphPattern::Filter { expr, inner } => {
                let b = self.pattern(inner)?;
                let b = self.plain(b);
                let (b, expr) = self.shallow(b, expr)?;
                let mut b = b;
                let conds = self.filter_conditions(&expr, &b.cols)?;
                b.wheres.extend(conds);
                Ok(b)
            }
            GraphPattern::Graph { name, inner } => {
                let saved = self.scope;
                self.scope = match name {
                    NamedNodePattern::NamedNode(n) if n.as_str() == HISTORY_GRAPH => {
                        GraphScope::History
                    }
                    NamedNodePattern::NamedNode(n) => {
                        let id = self.constant_id(&n.clone().into())?;
                        GraphScope::Fixed(id)
                    }
                    NamedNodePattern::Variable(v) => GraphScope::Var(self.var(v)),
                };
                let accesses = self.aliases;
                let r = self.pattern(inner);
                let scope = self.scope;
                self.scope = saved;
                let mut b = r?;
                match scope {
                    GraphScope::Var(v) => {
                        if !b.cols.contains_key(&v) {
                            // No quad access binds the graph: enumerate named graphs.
                            let g = self.graph_list_block(v);
                            b = self.join(b, g)?;
                        }
                    }
                    GraphScope::Fixed(id) => {
                        if self.aliases == accesses || !has_quad_access(p) {
                            // GRAPH <g> {} only matches if the graph exists.
                            b = self.plain(b);
                            b.wheres
                                .push(format!("EXISTS (SELECT 1 FROM graphs WHERE id = {id})"));
                        }
                    }
                    GraphScope::Default | GraphScope::History => {}
                }
                Ok(b)
            }
            GraphPattern::Extend {
                inner,
                variable,
                expression,
            } => {
                let b = self.pattern(inner)?;
                let mut b = self.plain(b);
                let restore = self.join_terms(&mut b, expression);
                let (b, expression) = self.shallow(b, expression)?;
                let mut b = b;
                let expression = &expression;
                let v = self.expr_term(expression, &b.cols)?;
                Self::restore_terms(&mut b, restore);
                let idx = self.var(variable);
                let (nullable, computed) = match expression {
                    Expression::Variable(x) => {
                        let xi = self.var(x);
                        (b.cols.get(&xi).is_none_or(|b| b.nullable), false)
                    }
                    Expression::NamedNode(_) | Expression::Literal(_) => (false, true),
                    _ => (true, true),
                };
                let col = match &v.id {
                    Some(id) if v.decodable => Col::Id(id.clone()),
                    _ => Col::Val(Box::new(v)),
                };
                b.cols.insert(
                    idx,
                    Binding {
                        col,
                        nullable,
                        computed,
                        correlated: false,
                    },
                );
                Ok(b)
            }
            GraphPattern::Values {
                variables,
                bindings,
            } => self.values(variables, bindings),
            GraphPattern::OrderBy { inner, expression } => {
                let mut b = self.pattern(inner)?;
                // Sort keys over an aggregate of this block would put the aggregate inside
                // the keys' scalar subqueries (a SQLite error): sort a sealed block instead.
                let reads_aggregate = b.stage == Stage::Grouped
                    && expression.iter().any(|oe| {
                        let (OrderExpression::Asc(e) | OrderExpression::Desc(e)) = oe;
                        expr_mentions(e, &mut |v| {
                            self.vars
                                .get(v)
                                .and_then(|i| b.cols.get(i))
                                .is_some_and(|c| c.computed)
                        })
                    });
                if b.stage > Stage::Grouped || reads_aggregate {
                    b = self.seal(b);
                }
                for oe in expression {
                    let (e, desc) = match oe {
                        OrderExpression::Asc(e) => (e, false),
                        OrderExpression::Desc(e) => (e, true),
                    };
                    let v = self.expr_term(e, &b.cols)?;
                    for k in order_keys(&v) {
                        b.order_by.push(if desc { format!("{k} DESC") } else { k });
                    }
                }
                b.stage = Stage::Ordered;
                Ok(b)
            }
            GraphPattern::Project { inner, variables } => {
                // Inside `GRAPH ?g`, a subquery's own ?g (if not projected) is a different
                // variable: bind the graph to a hidden variable and expose it as ?g after
                // projection.
                let graph_var = match self.scope {
                    GraphScope::Var(g) => Some(g),
                    _ => None,
                };
                let hidden = graph_var.map(|_| self.fresh_var("g"));
                let saved = self.scope;
                if let Some(h) = hidden {
                    self.scope = GraphScope::Var(h);
                }
                let r = self.pattern(inner);
                self.scope = saved;
                let mut b = r?;
                if b.stage >= Stage::Distinct {
                    b = self.seal(b);
                }
                let hidden_binding = hidden.and_then(|h| b.cols.get(&h).cloned());
                let mut cols = BTreeMap::new();
                for v in variables {
                    let idx = self.var(v);
                    let bind = b.cols.remove(&idx).unwrap_or(Binding {
                        col: Col::Id("NULL".into()),
                        nullable: true,
                        computed: true,
                        correlated: false,
                    });
                    cols.insert(idx, bind);
                }
                b.cols = cols;
                if let (Some(g), Some(h)) = (graph_var, hidden_binding) {
                    match b.cols.get(&g).and_then(|x| x.col.key().map(str::to_string)) {
                        Some(inner_g) => {
                            let hk = h.col.key().unwrap_or("NULL").to_string();
                            b.wheres.push(format!("{inner_g} = {hk}"));
                        }
                        None => {
                            b.cols.insert(g, h);
                        }
                    }
                }
                Ok(b)
            }
            GraphPattern::Distinct { inner } => {
                let mut b = self.pattern(inner)?;
                if b.stage >= Stage::Distinct {
                    b = self.seal(b);
                }
                b.distinct = true;
                b.stage = Stage::Distinct;
                Ok(b)
            }
            // REDUCED may drop any duplicates; Oxigraph drops them all, so do we.
            GraphPattern::Reduced { inner } => self.pattern(&GraphPattern::Distinct {
                inner: inner.clone(),
            }),
            GraphPattern::Slice {
                inner,
                start,
                length,
            } => {
                let mut b = self.pattern(inner)?;
                if b.stage == Stage::Sliced {
                    b = self.seal(b);
                }
                b.offset = *start;
                b.limit = *length;
                b.stage = Stage::Sliced;
                Ok(b)
            }
            other => self.pattern_ext(other),
        }
    }

    /// OPTIONAL, UNION, MINUS, GROUP BY and property paths live in `ops.rs`.
    fn pattern_ext(&mut self, p: &GraphPattern) -> Result<Block> {
        self.pattern_m2(p)
    }

    /// Condition that two bindings of the same variable agree (SPARQL compatibility), and the
    /// binding that results from merging them.
    pub(crate) fn unify(a: &Binding, b: &Binding) -> (String, Binding) {
        let eq = match (a.col.key(), b.col.key()) {
            (Some(x), Some(y)) => format!("{x} = {y}"),
            _ => expr::same_term(&a.col.value(), &b.col.value()),
        };
        if !a.nullable && !b.nullable {
            return (eq, a.clone());
        }
        let is_null = |x: &Binding| match &x.col {
            Col::Id(i) => format!("{i} IS NULL"),
            Col::Val(v) => format!("({}) IS NULL", v.kind),
        };
        let cond = format!("({eq} OR {} OR {})", is_null(a), is_null(b));
        let col = match (&a.col, &b.col) {
            (Col::Id(x), Col::Id(y)) => Col::Id(format!("COALESCE({x}, {y})")),
            _ => {
                let (av, bv) = (a.col.value(), b.col.value());
                Col::Val(Box::new(Self::choose(
                    &format!("(({}) IS NOT NULL)", av.kind),
                    &av,
                    &bv,
                )))
            }
        };
        (
            cond,
            Binding {
                col,
                nullable: a.nullable && b.nullable,
                computed: true,
                correlated: a.correlated || b.correlated,
            },
        )
    }

    fn graph_list_block(&mut self, v: usize) -> Block {
        let a = self.alias("gr");
        let mut b = Block::default();
        b.from.push(FromItem {
            join: Join::First,
            item: format!("graphs {a}"),
        });
        if let Some(named) = &self.dataset.named {
            b.wheres.push(in_list(&format!("{a}.id"), named));
        }
        b.cols.insert(v, Binding::id(format!("{a}.id")));
        b
    }

    fn pos(&mut self, t: &TermPattern, bnodes: &mut HashMap<String, usize>) -> Result<Pos> {
        Ok(match t {
            TermPattern::NamedNode(n) => Pos::Const(self.constant_id(&n.clone().into())?),
            TermPattern::Literal(l) => Pos::Const(self.constant_id(&l.clone().into())?),
            TermPattern::Variable(v) => Pos::Var(self.var(v)),
            TermPattern::BlankNode(b) => {
                // Blank nodes act as variables; the parser gives them query-unique labels, so
                // one hidden variable per label keeps them joined across paths and BGPs.
                let _ = bnodes;
                Pos::Var(self.var(&Variable::new_unchecked(format!("\u{1}b{}", b.as_str()))))
            }
            TermPattern::Triple(tp) => {
                if let Some(t) = ground_triple(tp) {
                    Pos::Const(self.constant_id(&t.into())?)
                } else {
                    let v = self.fresh_var("t");
                    self.pending_triples.push((v, (**tp).clone()));
                    Pos::Var(v)
                }
            }
        })
    }

    fn pos_nn(&mut self, p: &NamedNodePattern) -> Result<Pos> {
        Ok(match p {
            NamedNodePattern::NamedNode(n) => Pos::Const(self.constant_id(&n.clone().into())?),
            NamedNodePattern::Variable(v) => Pos::Var(self.var(v)),
        })
    }

    /// Binds a SQL column to a pattern position.
    pub(crate) fn bind_pos(&mut self, b: &mut Block, colsql: &str, pos: Pos) -> Result<()> {
        match pos {
            Pos::Const(id) => b.wheres.push(format!("{colsql} = {id}")),
            Pos::Var(v) => {
                if let Some(existing) = b.cols.get(&v) {
                    let cond = match existing.col.key() {
                        Some(x) => format!("{colsql} = {x}"),
                        None => expr::same_term(&V::from_id(colsql), &existing.col.value()),
                    };
                    b.wheres.push(if existing.nullable {
                        let null = match &existing.col {
                            Col::Id(x) => format!("{x} IS NULL"),
                            Col::Val(v) => format!("({}) IS NULL", v.kind),
                        };
                        format!("({null} OR {cond})")
                    } else {
                        cond
                    });
                    return Ok(());
                }
                let mut correlated = false;
                if let Some(outer) = self.outer_binding(v).cloned() {
                    let (cond, _) = Self::unify(&Binding::id(colsql), &outer);
                    b.wheres.push(cond);
                    correlated = true;
                }
                b.cols.insert(
                    v,
                    Binding {
                        correlated,
                        ..Binding::id(colsql)
                    },
                );
            }
        }
        Ok(())
    }

    /// Adds graph conditions for a quad alias according to the current scope and dataset.
    ///
    /// When another position is bound (`selective`), the graph equality is written as
    /// `+q.g = …`: the unary plus keeps SQLite from choosing the `gspo` index on a `g`
    /// equality that nearly every quad satisfies, so it uses the s/p/o permutation instead.
    pub(crate) fn graph_pos(&mut self, b: &mut Block, q: &str, selective: bool) -> Result<()> {
        let plus = if selective { "+" } else { "" };
        match self.scope {
            GraphScope::Default => match self.dataset.default.clone() {
                DefaultGraph::Zero => b.wheres.push(format!("{plus}{q}.g = {DEFAULT_GRAPH_ID}")),
                DefaultGraph::List(l) if l.is_empty() => b.wheres.push("0".into()),
                DefaultGraph::List(l) if l.len() == 1 => {
                    b.wheres.push(format!("{plus}{q}.g = {}", l[0]))
                }
                DefaultGraph::List(l) => {
                    b.wheres.push(in_list(&format!("{q}.g"), &l));
                    let d = self.alias("d");
                    b.wheres.push(format!(
                        "NOT EXISTS (SELECT 1 FROM {} {d} WHERE {d}.s = {q}.s AND {d}.p = {q}.p AND {d}.o = {q}.o AND {d}.g < {q}.g AND {})",
                        self.entailment().base(),
                        in_list(&format!("{d}.g"), &l)
                    ));
                }
                DefaultGraph::Union => {
                    let d = self.alias("d");
                    let base = self.entailment().base();
                    b.wheres.push(format!(
                        "NOT EXISTS (SELECT 1 FROM {base} {d} WHERE {d}.s = {q}.s AND {d}.p = {q}.p AND {d}.o = {q}.o AND {d}.g < {q}.g)"
                    ));
                }
            },
            GraphScope::Fixed(id) => {
                if self
                    .dataset
                    .named
                    .as_ref()
                    .is_some_and(|n| !n.contains(&id))
                {
                    b.wheres.push("0".into());
                }
                b.wheres.push(format!("{plus}{q}.g = {id}"));
            }
            GraphScope::Var(v) => {
                b.wheres.push(format!("{q}.g <> {DEFAULT_GRAPH_ID}"));
                if let Some(named) = self.dataset.named.clone() {
                    b.wheres.push(in_list(&format!("{q}.g"), &named));
                }
                self.bind_pos(b, &format!("{q}.g"), Pos::Var(v))?;
            }
            GraphScope::History => {}
        }
        Ok(())
    }

    /// `?c oxl:added <<( s p o )>>` / `oxl:removed` in the history graph: one row of the change
    /// log, the commit bound to its tick and the changed triple to the logged columns.
    fn change_access(&mut self, b: &mut Block, tp: &TriplePattern, added: bool) -> Result<()> {
        if self.stats.version.history == crate::version::History::None {
            return Err(Error::Other(
                "this store keeps no change log: raise its versioning level to `log` to read changes".into(),
            ));
        }
        let TermPattern::Triple(inner) = &tp.object else {
            return Err(Error::Other(
                "oxl:added and oxl:removed take a triple term: ?c oxl:added <<( ?s ?p ?o )>>"
                    .into(),
            ));
        };
        if matches!(inner.subject, TermPattern::Triple(_))
            || matches!(inner.object, TermPattern::Triple(_))
        {
            return Err(Error::unsupported(
                "nested triple terms in a change pattern",
            ));
        }
        let mut bnodes = HashMap::new();
        let c = self.pos(&tp.subject, &mut bnodes)?;
        let (s, p, o) = (
            self.pos(&inner.subject, &mut bnodes)?,
            self.pos_nn(&inner.predicate)?,
            self.pos(&inner.object, &mut bnodes)?,
        );
        let l = self.alias("hl");
        b.from.push(FromItem {
            join: if b.from.is_empty() {
                Join::First
            } else {
                Join::Inner
            },
            item: format!("quad_log {l}"),
        });
        b.wheres.push(format!("{l}.op = {}", u8::from(added)));
        self.bind_pos(b, &format!("({l}.tx + {})", expr::INT_BASE), c)?;
        self.bind_pos(b, &format!("{l}.s"), s)?;
        self.bind_pos(b, &format!("{l}.p"), p)?;
        self.bind_pos(b, &format!("{l}.o"), o)?;
        Ok(())
    }

    /// Builds entailed-triple sources for this query's reasoning options.
    pub(crate) fn entailment(&self) -> Entailment<'a> {
        Entailment {
            reasoning: self.options.reasoning,
            inferred: self.options.include_inferred,
            hide_schema: !self.options.include_schema_graphs,
            transitive: &self.stats.transitive,
            scopes: &self.stats.schema_scopes,
            max_compound: self.caps.max_compound_select,
            as_of: self.as_of,
        }
    }

    /// Adds a `quads` alias bound to the three positions (in the current graph scope).
    pub(crate) fn quad_access(
        &mut self,
        b: &mut Block,
        join: Join,
        s: Pos,
        p: Pos,
        o: Pos,
    ) -> Result<String> {
        let q = self.alias("q");
        let bound = |pos: Pos, b: &Block, me: &Self| match pos {
            Pos::Const(_) => true,
            Pos::Var(v) => b.cols.contains_key(&v) || me.outer_binding(v).is_some(),
        };
        let selective = bound(s, b, self) || bound(p, b, self) || bound(o, b, self);
        let ent = self.entailment();
        // With reasoning, a merged default graph is merged inside the derived table (so an
        // entailed triple appears once), which then only exposes graph 0.
        let merge = match (&self.scope, &self.dataset.default) {
            (GraphScope::Default, DefaultGraph::Union) if ent.active() => {
                Some(GraphFilter::Merge(None))
            }
            (GraphScope::Default, DefaultGraph::List(l)) if ent.active() && l.len() > 1 => {
                Some(GraphFilter::Merge(Some(l.clone())))
            }
            _ => None,
        };
        let source = if matches!(self.scope, GraphScope::History) {
            crate::version::history_sql(&self.stats.version)?
        } else if ent.active() {
            let c = |p: Pos| match p {
                Pos::Const(id) => Some(id),
                Pos::Var(_) => None,
            };
            ent.source(
                c(s),
                c(p),
                c(o),
                merge.as_ref().unwrap_or(&GraphFilter::Keep),
            )
        } else {
            ent.base()
        };
        b.from.push(FromItem {
            join: if b.from.is_empty() { Join::First } else { join },
            item: format!("{source} {q}"),
        });
        self.bind_pos(b, &format!("{q}.s"), s)?;
        self.bind_pos(b, &format!("{q}.p"), p)?;
        self.bind_pos(b, &format!("{q}.o"), o)?;
        if merge.is_none() {
            self.graph_pos(b, &q, selective)?;
        }
        Ok(q)
    }

    fn bgp(&mut self, patterns: &[TriplePattern]) -> Result<Block> {
        if patterns.is_empty() {
            return Ok(Block::default());
        }
        if matches!(self.scope, GraphScope::History) {
            let change = |tp: &TriplePattern| match &tp.predicate {
                NamedNodePattern::NamedNode(n) if n.as_str() == crate::version::vocab::ADDED => {
                    Some(true)
                }
                NamedNodePattern::NamedNode(n) if n.as_str() == crate::version::vocab::REMOVED => {
                    Some(false)
                }
                _ => None,
            };
            if patterns.iter().any(|tp| change(tp).is_some()) {
                let rest: Vec<TriplePattern> = patterns
                    .iter()
                    .filter(|tp| change(tp).is_none())
                    .cloned()
                    .collect();
                let mut b = self.bgp(&rest)?;
                for tp in patterns {
                    if let Some(added) = change(tp) {
                        self.change_access(&mut b, tp, added)?;
                    }
                }
                return Ok(b);
            }
        }
        let mut bnodes = HashMap::new();
        let mut enc = Vec::with_capacity(patterns.len());
        for tp in patterns {
            enc.push([
                self.pos(&tp.subject, &mut bnodes)?,
                self.pos_nn(&tp.predicate)?,
                self.pos(&tp.object, &mut bnodes)?,
            ]);
        }
        let pre: HashSet<usize> = self
            .outer
            .iter()
            .flat_map(|m| m.keys().copied())
            .chain(self.plan_hint.iter().flatten().copied())
            .collect();
        let ord: Vec<usize> = if self.options.sqlite_planner {
            (0..enc.len()).collect()
        } else {
            plan::order(&enc, &pre, self.stats)
        };
        let join = if self.options.sqlite_planner {
            Join::Inner
        } else {
            Join::Cross
        };
        // Record the plan for explain(): order, estimated rows, Cartesian products.
        {
            let mut bound = pre.clone();
            let mut steps = Vec::new();
            for (n, i) in ord.iter().enumerate() {
                let est = plan::estimate(&enc[*i], &bound, self.stats);
                let vars: Vec<usize> = enc[*i]
                    .iter()
                    .filter_map(|p| match p {
                        Pos::Var(v) => Some(*v),
                        Pos::Const(_) => None,
                    })
                    .collect();
                if n > 0 && !vars.iter().any(|v| bound.contains(v)) {
                    self.notes.push(format!(
                        "warning: Cartesian product: triple pattern {} shares no variable with the patterns before it",
                        patterns[*i]
                    ));
                }
                steps.push(format!("{} (~{est:.0} rows)", patterns[*i]));
                bound.extend(vars);
            }
            self.notes.push(format!(
                "join order ({}): {}",
                if self.options.sqlite_planner {
                    "SQLite planner"
                } else if self.stats.available {
                    "statistics"
                } else {
                    "heuristics, run optimize() for statistics"
                },
                steps.join(" → ")
            ));
        }
        let mut b = Block::default();
        for i in ord {
            let [s, p, o] = enc[i];
            self.quad_access(&mut b, join.clone(), s, p, o)?;
        }
        // RDF 1.2 triple-term patterns: join `triple_terms` on the term's id and match its
        // components (possibly nested).
        while let Some((v, tp)) = self.pending_triples.pop() {
            let Some(key) = b.cols.get(&v).and_then(|x| x.col.key().map(str::to_string)) else {
                return Err(Error::unsupported("unbound triple term pattern"));
            };
            let t = self.alias("tt");
            b.from.push(FromItem {
                join: Join::Inner,
                item: format!("triple_terms {t}"),
            });
            b.wheres.push(format!("{t}.id = {key}"));
            let sp = self.pos(&tp.subject, &mut bnodes)?;
            let pp = self.pos_nn(&tp.predicate)?;
            let op = self.pos(&tp.object, &mut bnodes)?;
            self.bind_pos(&mut b, &format!("{t}.s"), sp)?;
            self.bind_pos(&mut b, &format!("{t}.p"), pp)?;
            self.bind_pos(&mut b, &format!("{t}.o"), op)?;
        }
        Ok(b)
    }

    /// Joins two blocks (SPARQL Join).
    pub(crate) fn join(&mut self, a: Block, b: Block) -> Result<Block> {
        let mut a = self.plain(a);
        let b = self.plain(b);
        if a.is_unit() {
            return Ok(b);
        }
        if b.is_unit() {
            return Ok(a);
        }
        for (idx, bb) in b.cols {
            match a.cols.get(&idx).cloned() {
                None => {
                    a.cols.insert(idx, bb);
                }
                Some(ab) => {
                    let (cond, merged) = Self::unify(&ab, &bb);
                    a.wheres.push(cond);
                    a.cols.insert(idx, merged);
                }
            }
        }
        for mut item in b.from {
            if a.from.is_empty() {
                item.join = Join::First;
            } else if item.join == Join::First {
                item.join = Join::Inner;
            }
            a.from.push(item);
        }
        a.wheres.extend(b.wheres);
        Ok(a)
    }

    fn values(
        &mut self,
        variables: &[Variable],
        bindings: &[Vec<Option<GroundTerm>>],
    ) -> Result<Block> {
        let mut b = Block::default();
        if variables.is_empty() {
            match bindings.len() {
                0 => b.wheres.push("0".into()),
                1 => {}
                n => b.from.push(FromItem {
                    join: Join::First,
                    item: format!(
                        "(VALUES {}) AS {}",
                        vec!["(1)"; n].join(","),
                        self.alias("u")
                    ),
                }),
            }
            return Ok(b);
        }
        let idxs: Vec<usize> = variables.iter().map(|v| self.var(v)).collect();
        if bindings.is_empty() {
            b.wheres.push("0".into());
            for i in idxs {
                b.cols.insert(
                    i,
                    Binding {
                        nullable: true,
                        computed: true,
                        ..Binding::id("NULL")
                    },
                );
            }
            return Ok(b);
        }
        // Constants are not necessarily stored: each variable carries its full value (all
        // `V` fields) so no dictionary lookup is needed; the id remains the join key.
        let mut rows = Vec::new();
        let mut nullable = vec![false; idxs.len()];
        for row in bindings {
            let mut vals = Vec::new();
            for (j, t) in row.iter().enumerate() {
                let fields = match t {
                    None => {
                        nullable[j] = true;
                        val_fields(&V::null())
                    }
                    Some(t) => {
                        let term = ground_to_term(t);
                        let id = self.constant_id(&term)?;
                        val_fields(&V::from_term(&term, id)?)
                    }
                };
                vals.extend(fields);
            }
            rows.push(format!("({})", vals.join(",")));
        }
        let a = self.alias("vals");
        b.from.push(FromItem {
            join: Join::First,
            item: format!("(VALUES {}) AS {a}", rows.join(",")),
        });
        let n = VAL_FIELDS.len();
        for (j, i) in idxs.into_iter().enumerate() {
            let c = |k: usize| format!("{a}.column{}", j * n + k + 1);
            let v = V {
                id: Some(c(0)),
                kind: c(1),
                lex: c(2),
                dt: c(3),
                lang: c(4),
                num: c(5),
                nt: c(6),
                ts: c(7),
                boolv: c(8),
                stat: expr::Stat::Any,
                computed_num: false,
                decodable: false,
                aux: c(9),
                tz: "NULL".into(),
            };
            b.cols.insert(
                i,
                Binding {
                    col: Col::Val(Box::new(v)),
                    nullable: nullable[j],
                    computed: false,
                    correlated: false,
                },
            );
        }
        Ok(b)
    }

    /// Compiles a FILTER into WHERE conditions, keeping sargable equalities index-friendly.
    pub(crate) fn filter_conditions(
        &mut self,
        e: &Expression,
        cols: &BTreeMap<usize, Binding>,
    ) -> Result<Vec<String>> {
        if let Expression::And(a, b) = e {
            let mut out = self.filter_conditions(a, cols)?;
            out.extend(self.filter_conditions(b, cols)?);
            return Ok(out);
        }
        if let Expression::Equal(a, b) | Expression::SameTerm(a, b) = e {
            let pair = match (a.as_ref(), b.as_ref()) {
                (Expression::Variable(v), c) | (c, Expression::Variable(v)) => Some((v, c)),
                _ => None,
            };
            if let Some((v, c)) = pair {
                let t: Option<Term> = match c {
                    Expression::NamedNode(n) => Some(n.clone().into()),
                    Expression::Literal(l)
                        if matches!(e, Expression::SameTerm(..))
                            || l.language().is_some()
                            || l.datatype() == oxrdf::vocab::xsd::STRING =>
                    {
                        Some(l.clone().into())
                    }
                    _ => None,
                };
                if let Some(t) = t {
                    let idx = self.var(v);
                    let binding = cols.get(&idx).or_else(|| self.outer_binding(idx)).cloned();
                    if let Some(Binding {
                        col: Col::Id(x), ..
                    }) = binding
                    {
                        // In a FILTER an error and `false` both reject, so term equality is exact.
                        let id = self.constant_id(&t)?;
                        return Ok(vec![format!("{x} = {id}")]);
                    }
                }
            }
        }
        Ok(vec![self.expr_bool(e, cols)?])
    }
}

/// The triple term of a pattern without variables or blank nodes.
fn ground_triple(tp: &TriplePattern) -> Option<oxrdf::Triple> {
    fn term(t: &TermPattern) -> Option<Term> {
        Some(match t {
            TermPattern::NamedNode(n) => n.clone().into(),
            TermPattern::Literal(l) => l.clone().into(),
            TermPattern::Triple(tp) => ground_triple(tp)?.into(),
            TermPattern::BlankNode(_) | TermPattern::Variable(_) => return None,
        })
    }
    let s = match term(&tp.subject)? {
        Term::NamedNode(n) => oxrdf::NamedOrBlankNode::from(n),
        _ => return None,
    };
    let NamedNodePattern::NamedNode(p) = &tp.predicate else {
        return None;
    };
    Some(oxrdf::Triple::new(s, p.clone(), term(&tp.object)?))
}

fn has_quad_access(p: &GraphPattern) -> bool {
    match p {
        GraphPattern::Bgp { patterns } => !patterns.is_empty(),
        GraphPattern::Path { .. } => true,
        GraphPattern::Graph { .. } | GraphPattern::Values { .. } => false,
        GraphPattern::Join { left, right }
        | GraphPattern::LeftJoin { left, right, .. }
        | GraphPattern::Union { left, right }
        | GraphPattern::Minus { left, right } => has_quad_access(left) || has_quad_access(right),
        GraphPattern::Filter { inner, .. }
        | GraphPattern::Extend { inner, .. }
        | GraphPattern::OrderBy { inner, .. }
        | GraphPattern::Project { inner, .. }
        | GraphPattern::Distinct { inner }
        | GraphPattern::Reduced { inner }
        | GraphPattern::Slice { inner, .. }
        | GraphPattern::Group { inner, .. } => has_quad_access(inner),
        _ => true,
    }
}

/// Variables an expression reads (outside EXISTS patterns), each once.
fn expression_variables(e: &Expression, out: &mut Vec<Variable>) {
    match e {
        Expression::Variable(v) | Expression::Bound(v) if !out.contains(v) => out.push(v.clone()),
        Expression::Or(a, b)
        | Expression::And(a, b)
        | Expression::Equal(a, b)
        | Expression::SameTerm(a, b)
        | Expression::Greater(a, b)
        | Expression::GreaterOrEqual(a, b)
        | Expression::Less(a, b)
        | Expression::LessOrEqual(a, b)
        | Expression::Add(a, b)
        | Expression::Subtract(a, b)
        | Expression::Multiply(a, b)
        | Expression::Divide(a, b) => {
            expression_variables(a, out);
            expression_variables(b, out);
        }
        Expression::UnaryPlus(a) | Expression::UnaryMinus(a) | Expression::Not(a) => {
            expression_variables(a, out)
        }
        Expression::In(a, l) => {
            expression_variables(a, out);
            for x in l {
                expression_variables(x, out);
            }
        }
        Expression::If(a, b, c) => {
            for x in [a, b, c] {
                expression_variables(x, out);
            }
        }
        Expression::Coalesce(l) | Expression::FunctionCall(_, l) => {
            for x in l {
                expression_variables(x, out);
            }
        }
        _ => {}
    }
}

pub(crate) fn in_list(col: &str, ids: &[i64]) -> String {
    if ids.is_empty() {
        return "0".into();
    }
    format!(
        "{col} IN ({})",
        ids.iter()
            .map(ToString::to_string)
            .collect::<Vec<_>>()
            .join(",")
    )
}

pub(crate) fn ground_to_term(t: &GroundTerm) -> Term {
    match t {
        GroundTerm::NamedNode(n) => n.clone().into(),
        GroundTerm::Literal(l) => l.clone().into(),
        GroundTerm::Triple(t) => oxrdf::Triple::new(
            t.subject.clone(),
            t.predicate.clone(),
            ground_to_term(&t.object),
        )
        .into(),
    }
}

/// SQL ORDER BY keys implementing Oxigraph's ordering: unbound < blank nodes < IRIs <
/// literals < triple terms; numeric literals first by value, other literals by
/// (lexical form, datatype, language).
pub(crate) fn order_keys(v: &V) -> Vec<String> {
    use expr::{Stat, K_BNODE, K_IRI, K_TRIPLE};
    match v.stat {
        Stat::Numeric => vec![format!("({}) IS NOT NULL", v.kind), v.num.clone()],
        Stat::String => vec![format!("({}) IS NOT NULL", v.kind), v.lex.clone()],
        _ => vec![
            format!(
                "CASE WHEN ({k}) IS NULL THEN 0 WHEN ({k}) = {K_BNODE} THEN 1 WHEN ({k}) = {K_IRI} THEN 2 WHEN ({k}) = {K_TRIPLE} THEN 4 ELSE 3 END",
                k = v.kind
            ),
            format!("({}) IS NULL", v.num),
            v.num.clone(),
            v.lex.clone(),
            v.dt.clone(),
            v.lang.clone(),
        ]
        .into_iter()
        .chain(triple_order_keys(v))
        .collect(),
    }
}

/// Ordering key of triple terms: the sort key computed at write time.
fn triple_order_keys(v: &V) -> Vec<String> {
    use expr::K_TRIPLE;
    let (Some(id), true) = (&v.id, v.decodable) else {
        return Vec::new();
    };
    vec![format!(
        "(CASE WHEN ({}) = {K_TRIPLE} THEN (SELECT sk FROM triple_terms WHERE id = {id}) END)",
        v.kind
    )]
}

/// Whether `e` mentions a variable for which `f` holds (`EXISTS` counts as mentioning).
fn expr_mentions(e: &Expression, f: &mut impl FnMut(&Variable) -> bool) -> bool {
    match e {
        Expression::NamedNode(_) | Expression::Literal(_) => false,
        Expression::Variable(v) | Expression::Bound(v) => f(v),
        Expression::Or(a, b)
        | Expression::And(a, b)
        | Expression::Equal(a, b)
        | Expression::SameTerm(a, b)
        | Expression::Greater(a, b)
        | Expression::GreaterOrEqual(a, b)
        | Expression::Less(a, b)
        | Expression::LessOrEqual(a, b)
        | Expression::Add(a, b)
        | Expression::Subtract(a, b)
        | Expression::Multiply(a, b)
        | Expression::Divide(a, b) => expr_mentions(a, f) || expr_mentions(b, f),
        Expression::UnaryPlus(a) | Expression::UnaryMinus(a) | Expression::Not(a) => {
            expr_mentions(a, f)
        }
        Expression::In(a, l) => expr_mentions(a, f) || l.iter().any(|x| expr_mentions(x, f)),
        Expression::If(a, b, c) => {
            expr_mentions(a, f) || expr_mentions(b, f) || expr_mentions(c, f)
        }
        Expression::Coalesce(l) | Expression::FunctionCall(_, l) => {
            l.iter().any(|x| expr_mentions(x, f))
        }
        Expression::Exists(_) => true,
    }
}