oxilite_core/
query.rs

1//! Query compilation entry point, execution job and result decoding.
2//!
3// @lat: [[architecture#SPARQL to SQL compiler]]
4
5use crate::compiler::expr::format_number;
6use crate::compiler::{Block, Col, Compiler, QueryOptions, VAL_FIELDS};
7use crate::encoding::{blank_node_id, tag_of, Tag, DEFAULT_GRAPH_ID};
8use crate::error::{Error, Result};
9use crate::job::{Job, Step};
10use crate::resolve::{join, TermResolver};
11use crate::sql::{Capabilities, Request, Response, SqlValue, Statement};
12use crate::stats::Stats;
13use oxrdf::{BlankNode, Literal, NamedNode, NamedOrBlankNode, Term, Triple, Variable};
14use spargebra::algebra::GraphPattern;
15use spargebra::term::{NamedNodePattern, TermPattern, TriplePattern};
16use spargebra::Query;
17use std::collections::{HashMap, HashSet};
18
19/// Decoded query results.
20#[derive(Debug, Clone, PartialEq)]
21pub enum QueryOutput {
22    Solutions {
23        variables: Vec<Variable>,
24        rows: Vec<Vec<Option<Term>>>,
25    },
26    Boolean(bool),
27    Graph(Vec<Triple>),
28}
29
30#[derive(Debug, Clone, Copy, PartialEq, Eq)]
31enum Out {
32    Id,
33    Val,
34}
35
36#[derive(Debug, Clone)]
37enum Form {
38    Select,
39    Ask,
40    Construct(Vec<TriplePattern>),
41    Describe,
42}
43
44/// A query compiled to SQL.
45#[derive(Debug, Clone)]
46pub struct CompiledQuery {
47    /// The main SQL statement.
48    pub sql: String,
49    form: Form,
50    variables: Vec<Variable>,
51    layout: Vec<Out>,
52    constants: HashMap<i64, Term>,
53    union_default_graph: bool,
54    /// The quad table DESCRIBE reads (the change log at a tick for as-of queries).
55    source: String,
56    /// Planner decisions and warnings.
57    pub notes: Vec<String>,
58}
59
60impl CompiledQuery {
61    /// Human-readable description for `explain()`: the SQL, join orders with estimated
62    /// cardinalities, and warnings.
63    pub fn explain(&self) -> String {
64        let mut out = format!(
65            "-- oxilite: fully compiled to SQL ({} output variable(s))\n",
66            self.variables.len()
67        );
68        for n in &self.notes {
69            out.push_str("-- ");
70            out.push_str(n);
71            out.push('\n');
72        }
73        out.push_str(&self.sql);
74        out
75    }
76}
77
78fn projected(p: &GraphPattern) -> Option<&[Variable]> {
79    match p {
80        GraphPattern::Project { variables, .. } => Some(variables),
81        GraphPattern::Slice { inner, .. }
82        | GraphPattern::Distinct { inner }
83        | GraphPattern::Reduced { inner }
84        | GraphPattern::OrderBy { inner, .. } => projected(inner),
85        _ => None,
86    }
87}
88
89fn template_vars(template: &[TriplePattern], out: &mut Vec<Variable>) {
90    fn tp(t: &TermPattern, out: &mut Vec<Variable>) {
91        match t {
92            TermPattern::Variable(v) if !out.contains(v) => out.push(v.clone()),
93            TermPattern::Triple(t) => {
94                tp(&t.subject, out);
95                if let NamedNodePattern::Variable(v) = &t.predicate {
96                    if !out.contains(v) {
97                        out.push(v.clone());
98                    }
99                }
100                tp(&t.object, out);
101            }
102            _ => {}
103        }
104    }
105    for t in template {
106        tp(&t.subject, out);
107        if let NamedNodePattern::Variable(v) = &t.predicate {
108            if !out.contains(v) {
109                out.push(v.clone());
110            }
111        }
112        tp(&t.object, out);
113    }
114}
115
116/// Compiles a SPARQL query to SQL. Returns [`Error::Unsupported`] for anything that cannot
117/// be expressed in SQL on this backend.
118pub fn compile_query(
119    query: &Query,
120    stats: &Stats,
121    caps: &Capabilities,
122    options: &QueryOptions,
123) -> Result<CompiledQuery> {
124    let (dataset, pattern, base, form) = match query {
125        Query::Select {
126            dataset,
127            pattern,
128            base_iri,
129        } => (dataset, pattern, base_iri, Form::Select),
130        Query::Ask {
131            dataset,
132            pattern,
133            base_iri,
134        } => (dataset, pattern, base_iri, Form::Ask),
135        Query::Construct {
136            template,
137            dataset,
138            pattern,
139            base_iri,
140        } => (
141            dataset,
142            pattern,
143            base_iri,
144            Form::Construct(template.clone()),
145        ),
146        Query::Describe {
147            dataset,
148            pattern,
149            base_iri,
150        } => (dataset, pattern, base_iri, Form::Describe),
151    };
152    crate::version::check_query_options(stats, options)?;
153    let mut c = Compiler::new(
154        stats,
155        caps,
156        options,
157        dataset.as_ref(),
158        base.as_ref().map(|b| b.as_str().to_string()),
159    );
160    let block = c.pattern(pattern)?;
161    let variables: Vec<Variable> = match &form {
162        Form::Construct(t) => {
163            let mut v = Vec::new();
164            template_vars(t, &mut v);
165            v
166        }
167        _ => match projected(pattern) {
168            Some(v) => v.to_vec(),
169            None => block
170                .cols
171                .keys()
172                .map(|i| c.var_names[*i].clone())
173                .filter(|v| !v.as_str().starts_with('\u{1}'))
174                .collect(),
175        },
176    };
177    let idxs: Vec<usize> = variables.iter().map(|v| c.var(v)).collect();
178    let layout = idxs
179        .iter()
180        .map(|i| match block.cols.get(i).map(|b| &b.col) {
181            Some(Col::Val(_)) => Out::Val,
182            _ => Out::Id,
183        })
184        .collect();
185    let sql = match form {
186        Form::Ask => {
187            let inner = block.to_select(Some(&[]), false);
188            format!("SELECT EXISTS ({inner})")
189        }
190        _ => final_select(&block, &idxs, caps.int64_as_text),
191    };
192    if sql.len() > caps.max_sql_len {
193        return Err(Error::unsupported(format!(
194            "generated SQL is {} bytes, over the backend limit of {}",
195            sql.len(),
196            caps.max_sql_len
197        )));
198    }
199    Ok(CompiledQuery {
200        sql,
201        form,
202        variables,
203        layout,
204        constants: c.constants,
205        union_default_graph: options.union_default_graph,
206        source: options.as_of_tick.map_or_else(
207            || "quads".to_owned(),
208            |t| crate::version::as_of_sql(&t.to_string()),
209        ),
210        notes: c.notes,
211    })
212}
213
214fn final_select(block: &Block, idxs: &[usize], text_ids: bool) -> String {
215    block.to_select(Some(idxs), text_ids)
216}
217
218/// Decodes a computed value from its value columns.
219fn decode_val(cells: &[SqlValue]) -> Option<Term> {
220    let get = |name: &str| {
221        let i = VAL_FIELDS.iter().position(|f| *f == name).expect("field");
222        &cells[i]
223    };
224    let kind = get("k").as_i64()?;
225    let lex = get("l").clone().into_string();
226    let dt = get("d").clone().into_string();
227    let lang = get("g").clone().into_string();
228    let num = get("n").as_f64();
229    Some(match Tag::from_u8(kind as u8)? {
230        Tag::Iri => NamedNode::new(lex?).ok()?.into(),
231        Tag::BlankNode => BlankNode::new_unchecked(lex?).into(),
232        Tag::String => Literal::new_simple_literal(lex?).into(),
233        Tag::LangString => Literal::new_language_tagged_literal(lex?, lang?)
234            .ok()?
235            .into(),
236        Tag::DirLangString => {
237            let lang = lang?;
238            let (tag, dir) = lang.split_once("--")?;
239            let dir = if dir == "rtl" {
240                oxrdf::BaseDirection::Rtl
241            } else {
242                oxrdf::BaseDirection::Ltr
243            };
244            Literal::new_directional_language_tagged_literal(lex?, tag, dir)
245                .ok()?
246                .into()
247        }
248        Tag::Integer => match lex {
249            Some(l) => Literal::new_typed_literal(l, oxrdf::vocab::xsd::INTEGER).into(),
250            None => Literal::from(num? as i64).into(),
251        },
252        Tag::Boolean => Literal::from(get("b").as_i64()? != 0).into(),
253        Tag::Typed => {
254            let dt = dt?;
255            if let Some((sum, count)) = get("x")
256                .as_str()
257                .and_then(|a| a.strip_prefix("avg:"))
258                .and_then(|a| a.split_once('/'))
259            {
260                // Exact integer AVG: decimal division like Oxigraph.
261                let sum = oxsdatatypes::Decimal::from(sum.parse::<i64>().ok()?);
262                let count = oxsdatatypes::Decimal::from(count.parse::<i64>().ok()?);
263                return Some(Literal::from(sum.checked_div(count)?).into());
264            }
265            match lex {
266                Some(l) => Literal::new_typed_literal(l, NamedNode::new_unchecked(dt)).into(),
267                None => format_number(num?, &dt).into(),
268            }
269        }
270        Tag::Triple | Tag::Default => return None,
271    })
272}
273
274enum Cell {
275    Id(i64),
276    Term(Term),
277    Unbound,
278}
279
280enum State {
281    Start,
282    Main,
283    Resolving,
284    Describe,
285    DescribeResolving,
286}
287
288/// Runs a compiled query.
289pub struct QueryJob {
290    compiled: CompiledQuery,
291    caps: Capabilities,
292    state: State,
293    resolver: TermResolver,
294    cells: Vec<Vec<Cell>>,
295    describe_done: HashSet<i64>,
296    describe_quads: Vec<[i64; 3]>,
297}
298
299impl QueryJob {
300    pub fn new(compiled: CompiledQuery, caps: Capabilities) -> Self {
301        let resolver = TermResolver::with_constants(compiled.constants.clone());
302        Self {
303            compiled,
304            caps,
305            state: State::Start,
306            resolver,
307            cells: Vec::new(),
308            describe_done: HashSet::new(),
309            describe_quads: Vec::new(),
310        }
311    }
312
313    pub fn compiled(&self) -> &CompiledQuery {
314        &self.compiled
315    }
316
317    fn absorb_main(&mut self, response: Response) -> Result<()> {
318        let rs = response
319            .into_iter()
320            .next()
321            .ok_or_else(|| Error::backend("empty response"))?;
322        for row in rs.rows {
323            let mut out = Vec::with_capacity(self.compiled.layout.len());
324            let mut i = 0;
325            for l in &self.compiled.layout {
326                match l {
327                    Out::Id => {
328                        let cell = match row.get(i).and_then(SqlValue::as_i64) {
329                            Some(id) => {
330                                self.resolver.want(id);
331                                Cell::Id(id)
332                            }
333                            None => Cell::Unbound,
334                        };
335                        out.push(cell);
336                        i += 1;
337                    }
338                    Out::Val => {
339                        let cells = row
340                            .get(i..i + VAL_FIELDS.len())
341                            .ok_or_else(|| Error::corrupted("short row"))?;
342                        out.push(decode_val(cells).map_or(Cell::Unbound, Cell::Term));
343                        i += VAL_FIELDS.len();
344                    }
345                }
346            }
347            self.cells.push(out);
348        }
349        Ok(())
350    }
351
352    fn solutions(&self) -> Result<Vec<Vec<Option<Term>>>> {
353        self.cells
354            .iter()
355            .map(|row| {
356                row.iter()
357                    .map(|c| match c {
358                        Cell::Id(id) => self.resolver.get(*id).map(Some),
359                        Cell::Term(t) => Ok(Some(t.clone())),
360                        Cell::Unbound => Ok(None),
361                    })
362                    .collect()
363            })
364            .collect()
365    }
366
367    fn describe_request(&mut self, ids: Vec<i64>) -> Option<Request> {
368        let ids: Vec<i64> = ids
369            .into_iter()
370            .filter(|id| {
371                matches!(tag_of(*id), Some(Tag::Iri | Tag::BlankNode))
372                    && self.describe_done.insert(*id)
373            })
374            .collect();
375        if ids.is_empty() {
376            return None;
377        }
378        let c = |x: &str| {
379            if self.caps.int64_as_text {
380                format!("CAST({x} AS TEXT)")
381            } else {
382                x.into()
383            }
384        };
385        let graph = if self.compiled.union_default_graph {
386            String::new()
387        } else {
388            format!(" AND g = {DEFAULT_GRAPH_ID}")
389        };
390        let stmts = ids
391            .chunks(400)
392            .map(|chunk| {
393                Statement::new(format!(
394                    "SELECT DISTINCT {}, {}, {} FROM {} x WHERE s IN ({}){graph}",
395                    c("s"),
396                    c("p"),
397                    c("o"),
398                    self.compiled.source,
399                    join(chunk)
400                ))
401            })
402            .collect();
403        Some(Request::read(stmts))
404    }
405
406    fn finish(&mut self) -> Result<QueryOutput> {
407        let rows = self.solutions()?;
408        Ok(match &self.compiled.form {
409            Form::Select => QueryOutput::Solutions {
410                variables: self.compiled.variables.clone(),
411                rows,
412            },
413            Form::Ask => unreachable!("handled on the main response"),
414            Form::Construct(template) => {
415                let mut seen = HashSet::new();
416                let mut out = Vec::new();
417                for row in rows {
418                    let mut bnodes = HashMap::new();
419                    for t in template {
420                        if let Some(triple) =
421                            instantiate(t, &self.compiled.variables, &row, &mut bnodes)
422                        {
423                            if seen.insert(triple.clone()) {
424                                out.push(triple);
425                            }
426                        }
427                    }
428                }
429                QueryOutput::Graph(out)
430            }
431            Form::Describe => {
432                let mut out = Vec::new();
433                let mut seen = HashSet::new();
434                for [s, p, o] in &self.describe_quads {
435                    let (Ok(s), Ok(p), Ok(o)) = (
436                        self.resolver.get(*s),
437                        self.resolver.get(*p),
438                        self.resolver.get(*o),
439                    ) else {
440                        continue;
441                    };
442                    let (Ok(s), Term::NamedNode(p)) = (crate::encoding::to_subject(s), p) else {
443                        continue;
444                    };
445                    let t = Triple::new(s, p, o);
446                    if seen.insert(t.clone()) {
447                        out.push(t);
448                    }
449                }
450                QueryOutput::Graph(out)
451            }
452        })
453    }
454
455    fn after_resolution(&mut self) -> Result<Step<QueryOutput>> {
456        if let Some(r) = self.resolver.request(&self.caps) {
457            return Ok(Step::Execute(r));
458        }
459        if let (Form::Describe, State::Resolving) = (&self.compiled.form, &self.state) {
460            let ids: Vec<i64> = self
461                .cells
462                .iter()
463                .flatten()
464                .filter_map(|c| match c {
465                    Cell::Id(id) => Some(*id),
466                    Cell::Term(t) => match t {
467                        Term::NamedNode(n) => Some(crate::encoding::named_node_id(n.as_str())),
468                        Term::BlankNode(b) => Some(blank_node_id(b.as_str())),
469                        _ => None,
470                    },
471                    Cell::Unbound => None,
472                })
473                .collect();
474            self.state = State::Describe;
475            if let Some(r) = self.describe_request(ids) {
476                return Ok(Step::Execute(r));
477            }
478        }
479        Ok(Step::Done(self.finish()?))
480    }
481}
482
483fn instantiate(
484    t: &TriplePattern,
485    vars: &[Variable],
486    row: &[Option<Term>],
487    bnodes: &mut HashMap<String, BlankNode>,
488) -> Option<Triple> {
489    fn term(
490        t: &TermPattern,
491        vars: &[Variable],
492        row: &[Option<Term>],
493        bnodes: &mut HashMap<String, BlankNode>,
494    ) -> Option<Term> {
495        Some(match t {
496            TermPattern::NamedNode(n) => n.clone().into(),
497            TermPattern::Literal(l) => l.clone().into(),
498            TermPattern::BlankNode(b) => bnodes
499                .entry(b.as_str().to_string())
500                .or_default()
501                .clone()
502                .into(),
503            TermPattern::Variable(v) => row[vars.iter().position(|x| x == v)?].clone()?,
504            TermPattern::Triple(tp) => instantiate(tp, vars, row, bnodes)?.into(),
505        })
506    }
507    let s = match term(&t.subject, vars, row, bnodes)? {
508        Term::NamedNode(n) => NamedOrBlankNode::from(n),
509        Term::BlankNode(b) => NamedOrBlankNode::from(b),
510        _ => return None,
511    };
512    let p = match &t.predicate {
513        NamedNodePattern::NamedNode(n) => n.clone(),
514        NamedNodePattern::Variable(v) => match row[vars.iter().position(|x| x == v)?].clone()? {
515            Term::NamedNode(n) => n,
516            _ => return None,
517        },
518    };
519    let o = term(&t.object, vars, row, bnodes)?;
520    Some(Triple::new(s, p, o))
521}
522
523impl Job for QueryJob {
524    type Output = QueryOutput;
525
526    fn step(&mut self, response: Option<Response>) -> Result<Step<QueryOutput>> {
527        match (&self.state, response) {
528            (State::Start, _) => {
529                self.state = State::Main;
530                Ok(Step::Execute(Request::read(vec![Statement::new(
531                    self.compiled.sql.clone(),
532                )])))
533            }
534            (State::Main, Some(r)) => {
535                if let Form::Ask = self.compiled.form {
536                    let b = r
537                        .first()
538                        .and_then(|rs| rs.rows.first())
539                        .and_then(|row| row.first())
540                        .and_then(SqlValue::as_i64)
541                        .unwrap_or(0);
542                    return Ok(Step::Done(QueryOutput::Boolean(b != 0)));
543                }
544                self.absorb_main(r)?;
545                self.state = State::Resolving;
546                self.after_resolution()
547            }
548            (State::Resolving, Some(r)) => {
549                self.resolver.absorb(r)?;
550                self.after_resolution()
551            }
552            (State::Describe, Some(r)) => {
553                let mut next = Vec::new();
554                for rs in r {
555                    for row in rs.rows {
556                        let ids: Vec<i64> = row.iter().filter_map(SqlValue::as_i64).collect();
557                        if let [s, p, o] = ids[..] {
558                            self.resolver.want(s);
559                            self.resolver.want(p);
560                            self.resolver.want(o);
561                            self.describe_quads.push([s, p, o]);
562                            if tag_of(o) == Some(Tag::BlankNode) {
563                                next.push(o);
564                            }
565                        }
566                    }
567                }
568                // Concise bounded description: follow blank-node objects.
569                if let Some(r) = self.describe_request(next) {
570                    return Ok(Step::Execute(r));
571                }
572                self.state = State::DescribeResolving;
573                match self.resolver.request(&self.caps) {
574                    Some(r) => Ok(Step::Execute(r)),
575                    None => Ok(Step::Done(self.finish()?)),
576                }
577            }
578            (State::DescribeResolving, Some(r)) => {
579                self.resolver.absorb(r)?;
580                match self.resolver.request(&self.caps) {
581                    Some(r) => Ok(Step::Execute(r)),
582                    None => Ok(Step::Done(self.finish()?)),
583                }
584            }
585            (_, None) => Err(Error::Other("query job resumed without a response".into())),
586        }
587    }
588}
oxilite_core/query.rs

oxilite_core/
query.rs
