pyshacl_turbo_engine/
functions.rs

1//! Native SHACL functions and query preparation with per-evaluation registration.
2
3use std::collections::HashMap;
4use std::sync::{Arc, Mutex};
5
6use bigdecimal::BigDecimal;
7use oxigraph::model::*;
8use oxigraph::sparql::{
9    PreparedSparqlQuery, QueryEvaluationError, QueryResults, QuerySolutionIter, QueryTripleIter,
10    SparqlEvaluator,
11};
12use shacl_structural_model as model;
13use spargebra::{Query, SparqlParser, algebra::GraphPattern};
14
15use crate::graph::*;
16use crate::sparql_regex::RegexFunctions;
17use crate::{Error, Result};
18
19#[cfg(test)]
20#[path = "functions_tests.rs"]
21mod tests;
22
23pub(crate) struct NativeQuery {
24    pub parsed: Query,
25    prepared: PreparedSparqlQuery,
26    regex: RegexFunctions,
27}
28
29impl NativeQuery {
30    pub fn new(mut parsed: Query) -> Self {
31        let regex = RegexFunctions::lower(&mut parsed);
32        Self {
33            prepared: SparqlEvaluator::new().for_query(parsed.clone()),
34            parsed,
35            regex,
36        }
37    }
38
39    pub fn prepare(&self, functions: &Functions, data: &Dataset) -> NativePrepared {
40        if functions.0.is_empty() && !self.regex.required() {
41            NativePrepared {
42                query: self.prepared.clone(),
43                errors: None,
44            }
45        } else {
46            let errors = Arc::new(Mutex::new(None));
47            let evaluator = if functions.0.is_empty() {
48                SparqlEvaluator::new()
49            } else {
50                functions.evaluator(functions.snapshot(data), 0, errors.clone())
51            };
52            NativePrepared {
53                query: self
54                    .regex
55                    .register(evaluator, errors.clone())
56                    .for_query(self.parsed.clone()),
57                errors: Some(errors),
58            }
59        }
60    }
61}
62
63pub(crate) type FunctionErrors = Arc<Mutex<Option<Error>>>;
64
65pub(crate) struct NativePrepared {
66    query: PreparedSparqlQuery,
67    errors: Option<FunctionErrors>,
68}
69
70impl NativePrepared {
71    pub fn on_queryable_dataset(self, data: &Dataset) -> NativeBound<'_> {
72        NativeBound {
73            prepared: self,
74            data,
75            bindings: Vec::new(),
76        }
77    }
78}
79
80pub(crate) struct NativeBound<'a> {
81    prepared: NativePrepared,
82    data: &'a Dataset,
83    bindings: Vec<(Variable, Term)>,
84}
85
86impl<'a> NativeBound<'a> {
87    pub fn substitute_variable(mut self, variable: Variable, term: impl Into<Term>) -> Self {
88        self.bindings.push((variable, term.into()));
89        self
90    }
91
92    pub fn execute(self) -> Result<QueryResults<'a>> {
93        let mut query = self.prepared.query.on_queryable_dataset(self.data);
94        for (variable, value) in self.bindings {
95            query = query.substitute_variable(variable, value);
96        }
97        let result = query.execute()?;
98        let Some(errors) = self.prepared.errors else {
99            return Ok(result);
100        };
101        if let Some(error) = errors.lock().expect("function error mutex poisoned").take() {
102            return Err(error);
103        }
104        Ok(match result {
105            QueryResults::Solutions(rows) => QueryResults::Solutions(QuerySolutionIter::new(
106                rows.variables().to_vec().into(),
107                Checked {
108                    inner: rows,
109                    errors,
110                    done: false,
111                },
112            )),
113            QueryResults::Graph(triples) => QueryResults::Graph(QueryTripleIter::new(Checked {
114                inner: triples,
115                errors,
116                done: false,
117            })),
118            QueryResults::Boolean(value) => QueryResults::Boolean(value),
119        })
120    }
121}
122
123struct Checked<I> {
124    inner: I,
125    errors: FunctionErrors,
126    done: bool,
127}
128
129impl<T, I: Iterator<Item = std::result::Result<T, QueryEvaluationError>>> Iterator for Checked<I> {
130    type Item = std::result::Result<T, QueryEvaluationError>;
131    fn next(&mut self) -> Option<Self::Item> {
132        if self.done {
133            return None;
134        }
135        let next = self.inner.next();
136        if let Some(error) = self
137            .errors
138            .lock()
139            .expect("function error mutex poisoned")
140            .take()
141        {
142            self.done = true;
143            return Some(Err(QueryEvaluationError::Unexpected(Box::new(error))));
144        }
145        self.done = next.is_none();
146        next
147    }
148}
149
150pub(crate) struct Parameter {
151    pub path: NamedNode,
152    pub variable: Variable,
153    pub optional: bool,
154}
155
156pub(crate) fn read_parameters(graph: &Dataset, node: &Term) -> Result<Vec<model::Parameter<Term>>> {
157    values(graph, node, "parameter")?
158        .into_iter()
159        .map(|p| {
160            let path = crate::structural::iri_value(
161                one(graph, &p, "path")?
162                    .ok_or_else(|| Error::Invalid("SHACL parameter requires sh:path".into()))?,
163                "Parameter path",
164            )?;
165            let mut parameter = model::Parameter::new(p.clone(), path);
166            parameter.optional = boolean(graph, &p, "optional")?;
167            parameter.order = one(graph, &p, "order")?;
168            // Argument constraints are annotations, not executable shapes (ADR 0010).
169            // Their original triples remain in the source graph.
170            parameter.names = values(graph, &p, "name")?;
171            parameter.descriptions = values(graph, &p, "description")?;
172            Ok(parameter)
173        })
174        .collect()
175}
176
177pub(crate) fn lower_parameters(
178    declarations: Vec<model::Parameter<Term>>,
179) -> Result<Vec<Parameter>> {
180    let mut parameters = Vec::new();
181    for declaration in declarations {
182        let path = crate::structural::named_iri(declaration.path);
183        let name = path.as_str().rsplit(['#', '/']).next().unwrap();
184        let variable = Variable::new(name).map_err(|e| Error::Invalid(e.to_string()))?;
185        let order = declaration
186            .order
187            .map(|term| match term {
188                Term::Literal(l) => l
189                    .value()
190                    .parse::<BigDecimal>()
191                    .map_err(|e| Error::Invalid(e.to_string())),
192                _ => Err(Error::Invalid("Parameter order must be numeric".into())),
193            })
194            .transpose()?;
195        parameters.push((
196            order,
197            Parameter {
198                path,
199                variable,
200                optional: declaration.optional,
201            },
202        ));
203    }
204    let ordered = parameters.iter().all(|(o, _)| o.is_some());
205    parameters.sort_by(|(a, x), (b, y)| {
206        if ordered {
207            a.cmp(b)
208        } else {
209            x.variable.as_str().cmp(y.variable.as_str())
210        }
211    });
212    Ok(parameters.into_iter().map(|(_, p)| p).collect())
213}
214
215/// Make externally supplied variables available through the query's projection.
216pub(crate) fn project_bindings(query: &mut Query, bindings: &[Variable]) {
217    fn extend(pattern: &mut GraphPattern, bindings: &[Variable]) {
218        match pattern {
219            GraphPattern::Project { variables, .. } => {
220                for v in bindings {
221                    if !variables.contains(v) {
222                        variables.push(v.clone());
223                    }
224                }
225            }
226            GraphPattern::Distinct { inner }
227            | GraphPattern::Reduced { inner }
228            | GraphPattern::Slice { inner, .. }
229            | GraphPattern::OrderBy { inner, .. } => extend(inner, bindings),
230            _ => {
231                let mut variables = Vec::new();
232                pattern.on_in_scope_variable(|v| variables.push(v.clone()));
233                for v in bindings {
234                    if !variables.contains(v) {
235                        variables.push(v.clone());
236                    }
237                }
238                *pattern = GraphPattern::Project {
239                    inner: Box::new(pattern.clone()),
240                    variables,
241                };
242            }
243        }
244    }
245    match query {
246        Query::Select { pattern, .. }
247        | Query::Ask { pattern, .. }
248        | Query::Construct { pattern, .. }
249        | Query::Describe { pattern, .. } => extend(pattern, bindings),
250    }
251}
252
253struct Function {
254    query: NativeQuery,
255    parameters: Vec<Parameter>,
256    output: Option<Variable>,
257}
258
259#[derive(Clone, Default)]
260pub(crate) struct Functions(Arc<HashMap<NamedNode, Function>>, Option<Arc<Dataset>>);
261
262impl Functions {
263    /// Bind callback reads to one immutable rule/validation phase.
264    pub fn for_dataset(&self, data: &Dataset) -> Self {
265        Self(
266            self.0.clone(),
267            (!self.0.is_empty()).then(|| Arc::new(data.clone())),
268        )
269    }
270
271    fn snapshot(&self, data: &Dataset) -> Arc<Dataset> {
272        self.1.clone().unwrap_or_else(|| Arc::new(data.clone()))
273    }
274
275    pub fn compile(graph: &Dataset, namespaces: &[(String, String)]) -> Result<Self> {
276        let mut functions = HashMap::new();
277        for node in subjects(
278            graph,
279            &iri(RDF, "type"),
280            Some(&iri(SH, "SPARQLFunction").into()),
281        )? {
282            let name = crate::structural::iri_value(node.clone(), "SHACL function")?;
283            let select = one(graph, &node, "select")?;
284            let ask = one(graph, &node, "ask")?;
285            let query = match (select, ask) {
286                (Some(text @ Term::Literal(_)), None) => {
287                    model::SparqlSelectOrAsk::Select(crate::structural::query(graph, &node, text)?)
288                }
289                (None, Some(text @ Term::Literal(_))) => {
290                    model::SparqlSelectOrAsk::Ask(crate::structural::query(graph, &node, text)?)
291                }
292                _ => {
293                    return Err(Error::Invalid(format!(
294                        "Function {node} requires exactly one sh:select or sh:ask"
295                    )));
296                }
297            };
298            let mut declaration = model::ShaclFunction::new(name, query);
299            declaration.parameters = read_parameters(graph, &node)?;
300            let name = crate::structural::named_iri(declaration.id);
301            let parameters = lower_parameters(declaration.parameters)?;
302            let (query, is_ask) = match declaration.query {
303                model::SparqlSelectOrAsk::Select(query) => (query, false),
304                model::SparqlSelectOrAsk::Ask(query) => (query, true),
305            };
306            let Term::Literal(text) = query.text else {
307                unreachable!("Function query literal checked by reader")
308            };
309            let mut query = SparqlParser::new()
310                .parse_query(&format!(
311                    "{}\n{}",
312                    crate::sparql::prefixes(graph, query.prefixes, namespaces)?,
313                    text.value()
314                ))
315                .map_err(|e| Error::Invalid(format!("SPARQL function parse error: {e}")))?;
316            let output = match &query {
317                Query::Select { pattern, .. } if !is_ask => {
318                    let mut first = None;
319                    pattern.on_in_scope_variable(|v| {
320                        if first.is_none() {
321                            first = Some(v.clone());
322                        }
323                    });
324                    Some(first.ok_or_else(|| {
325                        Error::Invalid(format!("Function {node} must select a result variable"))
326                    })?)
327                }
328                Query::Ask { .. } if is_ask => None,
329                _ => {
330                    return Err(Error::Invalid(format!(
331                        "Function {node} query kind does not match its SHACL declaration"
332                    )));
333                }
334            };
335            project_bindings(
336                &mut query,
337                &parameters
338                    .iter()
339                    .map(|p| p.variable.clone())
340                    .collect::<Vec<_>>(),
341            );
342            functions.insert(
343                name,
344                Function {
345                    query: NativeQuery::new(query),
346                    parameters,
347                    output,
348                },
349            );
350        }
351        Ok(Self(Arc::new(functions), None))
352    }
353
354    pub fn contains(&self, name: &NamedNode) -> bool {
355        self.0.contains_key(name)
356    }
357
358    fn evaluator(
359        &self,
360        data: Arc<Dataset>,
361        depth: usize,
362        errors: FunctionErrors,
363    ) -> SparqlEvaluator {
364        let mut evaluator = SparqlEvaluator::new();
365        for name in self.0.keys() {
366            let (functions, data, function) = (self.clone(), data.clone(), name.clone());
367            let errors = errors.clone();
368            evaluator = evaluator.with_custom_function(name.clone(), move |args| {
369                if errors
370                    .lock()
371                    .expect("function error mutex poisoned")
372                    .is_some()
373                {
374                    return None;
375                }
376                match functions.call_inner(&function, args, data.clone(), depth + 1) {
377                    Ok(value) => value,
378                    Err(error) => {
379                        *errors.lock().expect("function error mutex poisoned") = Some(error);
380                        None
381                    }
382                }
383            });
384        }
385        evaluator
386    }
387
388    pub fn call(&self, name: &NamedNode, args: &[Term], data: &Dataset) -> Result<Option<Term>> {
389        self.call_inner(name, args, self.snapshot(data), 0)
390    }
391
392    fn call_inner(
393        &self,
394        name: &NamedNode,
395        args: &[Term],
396        data: Arc<Dataset>,
397        depth: usize,
398    ) -> Result<Option<Term>> {
399        if depth >= 64 {
400            return Err(Error::Runtime(
401                "SHACL function recursion limit exceeded".into(),
402            ));
403        }
404        let function = self
405            .0
406            .get(name)
407            .ok_or_else(|| Error::Invalid(format!("Unknown SHACL function {name}")))?;
408        if args.len() > function.parameters.len()
409            || function
410                .parameters
411                .iter()
412                .skip(args.len())
413                .any(|p| !p.optional)
414        {
415            return Err(Error::Invalid(format!(
416                "Wrong argument count for SHACL function {name}"
417            )));
418        }
419        let errors = Arc::new(Mutex::new(None));
420        let mut query = NativePrepared {
421            query: function
422                .query
423                .regex
424                .register(
425                    self.evaluator(data.clone(), depth, errors.clone()),
426                    errors.clone(),
427                )
428                .for_query(function.query.parsed.clone()),
429            errors: Some(errors),
430        }
431        .on_queryable_dataset(data.as_ref());
432        for (p, arg) in function.parameters.iter().zip(args) {
433            query = query.substitute_variable(p.variable.clone(), arg.clone());
434        }
435        match query.execute()? {
436            QueryResults::Boolean(value) => Ok(Some(Literal::from(value).into())),
437            QueryResults::Solutions(mut rows) => Ok(rows
438                .next()
439                .transpose()?
440                .and_then(|row| row.get(function.output.as_ref().unwrap()).cloned())),
441            _ => Err(Error::Invalid(
442                "SHACL function must return SELECT or ASK results".into(),
443            )),
444        }
445    }
446}
447
448/// Execute a query with SHACL functions declared in the supplied shapes graph.
449pub fn evaluate_query<'a>(
450    data: &'a Dataset,
451    shapes: &Dataset,
452    query: &str,
453    namespaces: &[(String, String)],
454) -> Result<QueryResults<'a>> {
455    let functions = Functions::compile(shapes, namespaces)?;
456    let parsed = SparqlParser::new()
457        .parse_query(query)
458        .map_err(|e| Error::Invalid(format!("SPARQL parse error: {e}")))?;
459    NativeQuery::new(parsed)
460        .prepare(&functions, data)
461        .on_queryable_dataset(data)
462        .execute()
463}
464
465#[cfg(test)]
466mod snapshot_tests {
467    use super::*;
468
469    #[test]
470    fn function_snapshot_is_shared_within_a_phase_and_refreshed_between_phases() {
471        let graph = crate::parse_graph(
472            "@prefix sh: <http://www.w3.org/ns/shacl#> . <urn:f> a sh:SPARQLFunction; sh:select \"SELECT (1 AS ?result) WHERE {}\" .",
473            oxigraph::io::RdfFormat::Turtle, None,
474        ).unwrap();
475        let functions = Functions::compile(&graph, &[]).unwrap();
476        let mut data = Dataset::new();
477        let phase = functions.for_dataset(&data);
478        let first = phase.snapshot(&data);
479        let again = phase.clone().snapshot(&data);
480        assert!(Arc::ptr_eq(&first, &again));
481        data.insert(
482            Quad::new(
483                NamedNode::new_unchecked("urn:s"),
484                NamedNode::new_unchecked("urn:p"),
485                Literal::from(1),
486                GraphName::DefaultGraph,
487            )
488            .as_ref(),
489        );
490        let next = functions.for_dataset(&data).snapshot(&data);
491        assert!(!Arc::ptr_eq(&first, &next));
492        assert_eq!(first.len(), 0);
493        assert_eq!(next.len(), 1);
494    }
495}
pyshacl_turbo_engine/functions.rs

pyshacl_turbo_engine/
functions.rs
