1#[cfg(feature = "sparql-12")]
2use crate::dataset::ExpressionTriple;
3use crate::dataset::{ExpressionTerm, InternalQuad, QueryableDataset};
4use crate::error::QueryEvaluationError;
5use crate::expression::{
6 CustomFunctionRegistry, ExpressionEvaluator, ExpressionEvaluatorContext, NumericBinaryOperands,
7 build_expression_evaluator, partial_cmp_literals, try_build_internal_expression_evaluator,
8};
9use crate::model::{QuerySolutionIter, QueryTripleIter};
10use crate::service::ServiceHandlerRegistry;
11use crate::{
12 AggregateFunctionAccumulator, CustomAggregateFunctionRegistry, QueryDatasetSpecification,
13};
14use json_event_parser::{JsonEvent, WriterJsonSerializer};
15use oxiri::Iri;
16#[cfg(feature = "sparql-12")]
17use oxrdf::{BaseDirection, NamedOrBlankNode};
18use oxrdf::{BlankNode, GraphName, Literal, NamedNode, Term, Triple, Variable};
19use oxsdatatypes::{DateTime, DayTimeDuration, Decimal, Double, Float, Integer};
20use rustc_hash::{FxBuildHasher, FxHashMap, FxHashSet, FxHasher};
21use spargebra::algebra::{AggregateFunction, PropertyPathExpression};
22#[cfg(feature = "sparql-12")]
23use spargebra::term::GroundTriple;
24use spargebra::term::{
25 GroundTerm, GroundTermPattern, NamedNodePattern, TermPattern, TriplePattern,
26};
27use sparopt::algebra::{
28 AggregateExpression, Expression, GraphPattern, JoinAlgorithm, LeftJoinAlgorithm,
29 MinusAlgorithm, OrderExpression,
30};
31use std::cell::Cell;
32use std::cmp::Ordering;
33use std::hash::{Hash, Hasher};
34use std::iter::{Peekable, empty, once};
35use std::marker::PhantomData;
36use std::rc::Rc;
37use std::sync::atomic::AtomicBool;
38use std::sync::{Arc, atomic};
39use std::{fmt, io};
40type InternalTupleEvaluator<'a, T> =
43 Rc<dyn Fn(InternalTuple<T>) -> InternalTuplesIterator<'a, T> + 'a>;
44
45struct EvalDataset<'a, D: QueryableDataset<'a>> {
47 dataset: Rc<D>,
48 specification: EncodedDatasetSpec<D::InternalTerm>,
49 cancellation_token: CancellationToken,
50 _lifetime: PhantomData<&'a ()>,
51}
52
53impl<'a, D: QueryableDataset<'a>> EvalDataset<'a, D> {
54 fn new(
55 dataset: D,
56 specification: QueryDatasetSpecification,
57 cancellation_token: CancellationToken,
58 ) -> Result<Self, QueryEvaluationError> {
59 let specification = EncodedDatasetSpec {
60 default: specification
61 .default
62 .map(|graph_names| {
63 graph_names
64 .into_iter()
65 .map(|graph_name| {
66 Ok(match graph_name {
67 GraphName::NamedNode(n) => {
68 Some(dataset.internalize_term(n.into())?)
69 }
70 GraphName::BlankNode(n) => {
71 Some(dataset.internalize_term(n.into())?)
72 }
73 GraphName::DefaultGraph => None,
74 })
75 })
76 .collect()
77 })
78 .transpose()
79 .map_err(|e: D::Error| QueryEvaluationError::Dataset(Box::new(e)))?,
80 named: specification
81 .named
82 .map(|graph_names| {
83 graph_names
84 .into_iter()
85 .map(|graph_name| dataset.internalize_term(graph_name.into()))
86 .collect()
87 })
88 .transpose()
89 .map_err(|e| QueryEvaluationError::Dataset(Box::new(e)))?,
90 };
91 Ok(Self {
92 dataset: Rc::new(dataset),
93 specification,
94 cancellation_token,
95 _lifetime: PhantomData,
96 })
97 }
98
99 fn underlying_internal_quads_for_pattern(
100 &self,
101 subject: Option<&D::InternalTerm>,
102 predicate: Option<&D::InternalTerm>,
103 object: Option<&D::InternalTerm>,
104 graph_name: Option<Option<&D::InternalTerm>>,
105 ) -> impl Iterator<Item = Result<InternalQuad<D::InternalTerm>, QueryEvaluationError>> + use<'a, D>
106 {
107 let cancellation_token = self.cancellation_token.clone();
108 self.dataset
109 .internal_quads_for_pattern(subject, predicate, object, graph_name)
110 .map(move |r| {
111 cancellation_token.ensure_alive()?;
112 r.map_err(|e| QueryEvaluationError::Dataset(Box::new(e)))
113 })
114 }
115
116 fn internal_quads_for_pattern(
117 &self,
118 subject: Option<&D::InternalTerm>,
119 predicate: Option<&D::InternalTerm>,
120 object: Option<&D::InternalTerm>,
121 graph_name: Option<Option<&D::InternalTerm>>,
122 ) -> Box<dyn Iterator<Item = Result<InternalQuad<D::InternalTerm>, QueryEvaluationError>> + 'a>
123 {
124 if let Some(graph_name) = graph_name {
125 if let Some(graph_name) = graph_name {
127 if self
129 .specification
130 .named
131 .as_ref()
132 .is_none_or(|d| d.contains(graph_name))
133 {
134 Box::new(self.underlying_internal_quads_for_pattern(
136 subject,
137 predicate,
138 object,
139 Some(Some(graph_name)),
140 ))
141 } else {
142 Box::new(empty())
143 }
144 } else if let Some(default_graph_graphs) = &self.specification.default {
145 if default_graph_graphs.len() == 1 {
147 Box::new(
149 self.underlying_internal_quads_for_pattern(
150 subject,
151 predicate,
152 object,
153 Some(default_graph_graphs[0].as_ref()),
154 )
155 .map(|quad| {
156 let mut quad = quad?;
157 quad.graph_name = None;
158 Ok(quad)
159 }),
160 )
161 } else {
162 let iters = default_graph_graphs
163 .iter()
164 .map(|graph_name| {
165 self.underlying_internal_quads_for_pattern(
166 subject,
167 predicate,
168 object,
169 Some(graph_name.as_ref()),
170 )
171 })
172 .collect::<Vec<_>>();
173 Box::new(iters.into_iter().flatten().map(|quad| {
174 let mut quad = quad?;
175 quad.graph_name = None;
176 Ok(quad)
177 }))
178 }
179 } else {
180 Box::new(
182 self.underlying_internal_quads_for_pattern(subject, predicate, object, None)
183 .map(|quad| {
184 let mut quad = quad?;
185 quad.graph_name = None;
186 Ok(quad)
187 }),
188 )
189 }
190 } else if let Some(named_graphs) = &self.specification.named {
191 let iters = named_graphs
193 .iter()
194 .map(|graph_name| {
195 self.underlying_internal_quads_for_pattern(
196 subject,
197 predicate,
198 object,
199 Some(Some(graph_name)),
200 )
201 })
202 .collect::<Vec<_>>();
203 Box::new(iters.into_iter().flatten())
204 } else {
205 Box::new(
207 self.underlying_internal_quads_for_pattern(subject, predicate, object, None)
208 .filter(|q| !q.as_ref().is_ok_and(|q| q.graph_name.is_none())),
209 )
210 }
211 }
212
213 fn internal_named_graphs(
214 &self,
215 ) -> Box<dyn Iterator<Item = Result<D::InternalTerm, QueryEvaluationError>> + 'a> {
216 if let Some(named_graphs) = &self.specification.named {
217 Box::new(
218 named_graphs
219 .iter()
220 .cloned()
221 .map(Ok)
222 .collect::<Vec<_>>()
223 .into_iter(),
224 )
225 } else {
226 let cancellation_token = self.cancellation_token.clone();
227 Box::new(self.dataset.internal_named_graphs().map(move |r| {
228 cancellation_token.ensure_alive()?;
229 r.map_err(|e| QueryEvaluationError::Dataset(Box::new(e)))
230 }))
231 }
232 }
233
234 fn contains_internal_graph_name(
235 &self,
236 graph_name: &D::InternalTerm,
237 ) -> Result<bool, QueryEvaluationError> {
238 if let Some(named_graphs) = &self.specification.named {
239 Ok(named_graphs.contains(graph_name))
240 } else {
241 self.dataset
242 .contains_internal_graph_name(graph_name)
243 .map_err(|e| QueryEvaluationError::Dataset(Box::new(e)))
244 }
245 }
246
247 fn internalize_term(&self, term: Term) -> Result<D::InternalTerm, QueryEvaluationError> {
248 self.cancellation_token.ensure_alive()?;
249 self.dataset
250 .internalize_term(term)
251 .map_err(|e| QueryEvaluationError::Dataset(Box::new(e)))
252 }
253
254 fn externalize_term(&self, term: D::InternalTerm) -> Result<Term, QueryEvaluationError> {
255 self.dataset
256 .externalize_term(term)
257 .map_err(|e| QueryEvaluationError::Dataset(Box::new(e)))
258 }
259
260 fn externalize_expression_term(
261 &self,
262 term: D::InternalTerm,
263 ) -> Result<ExpressionTerm, QueryEvaluationError> {
264 self.dataset
265 .externalize_expression_term(term)
266 .map_err(|e| QueryEvaluationError::Dataset(Box::new(e)))
267 }
268
269 fn internalize_expression_term(
270 &self,
271 term: ExpressionTerm,
272 ) -> Result<D::InternalTerm, QueryEvaluationError> {
273 self.dataset
274 .internalize_expression_term(term)
275 .map_err(|e| QueryEvaluationError::Dataset(Box::new(e)))
276 }
277
278 fn internal_term_effective_boolean_value(
279 &self,
280 term: D::InternalTerm,
281 ) -> Result<Option<bool>, QueryEvaluationError> {
282 self.dataset
283 .internal_term_effective_boolean_value(term)
284 .map_err(|e| QueryEvaluationError::Dataset(Box::new(e)))
285 }
286}
287
288impl<'a, D: QueryableDataset<'a>> Clone for EvalDataset<'a, D> {
289 #[inline]
290 fn clone(&self) -> Self {
291 Self {
292 dataset: Rc::clone(&self.dataset),
293 specification: self.specification.clone(),
294 cancellation_token: self.cancellation_token.clone(),
295 _lifetime: self._lifetime,
296 }
297 }
298}
299
300#[derive(Clone)]
301struct EncodedDatasetSpec<T> {
302 default: Option<Vec<Option<T>>>,
303 named: Option<Vec<T>>,
304}
305
306pub struct InternalTuple<T> {
307 inner: Vec<Option<T>>,
308}
309
310impl<T> InternalTuple<T> {
311 pub fn with_capacity(capacity: usize) -> Self {
312 Self {
313 inner: Vec::with_capacity(capacity),
314 }
315 }
316
317 pub fn capacity(&self) -> usize {
318 self.inner.capacity()
319 }
320
321 pub fn contains(&self, index: usize) -> bool {
322 self.inner.get(index).is_some_and(Option::is_some)
323 }
324
325 pub fn get(&self, index: usize) -> Option<&T> {
326 self.inner.get(index).unwrap_or(&None).as_ref()
327 }
328}
329
330impl<T: Clone> InternalTuple<T> {
331 pub fn iter(&self) -> impl Iterator<Item = Option<T>> + '_ {
332 self.inner.iter().cloned()
333 }
334
335 pub fn set(&mut self, index: usize, value: T) {
336 if self.inner.len() <= index {
337 self.inner.resize(index + 1, None);
338 }
339 self.inner[index] = Some(value);
340 }
341}
342
343impl<T: Clone + Eq> InternalTuple<T> {
344 pub fn combine_with(&self, other: &Self) -> Option<Self> {
345 if self.inner.len() < other.inner.len() {
346 let mut result = other.inner.clone();
347 for (key, self_value) in self.inner.iter().enumerate() {
348 if let Some(self_value) = self_value {
349 match &other.inner[key] {
350 Some(other_value) => {
351 if self_value != other_value {
352 return None;
353 }
354 }
355 None => result[key] = Some(self_value.clone()),
356 }
357 }
358 }
359 Some(Self { inner: result })
360 } else {
361 let mut result = self.inner.clone();
362 for (key, other_value) in other.inner.iter().enumerate() {
363 if let Some(other_value) = other_value {
364 match &self.inner[key] {
365 Some(self_value) => {
366 if self_value != other_value {
367 return None;
368 }
369 }
370 None => result[key] = Some(other_value.clone()),
371 }
372 }
373 }
374 Some(Self { inner: result })
375 }
376 }
377}
378
379impl<T: Clone> Clone for InternalTuple<T> {
380 fn clone(&self) -> Self {
381 Self {
382 inner: self.inner.clone(),
383 }
384 }
385}
386
387impl<T: Eq> PartialEq for InternalTuple<T> {
388 #[inline]
389 fn eq(&self, other: &InternalTuple<T>) -> bool {
390 self.inner == other.inner
391 }
392}
393
394impl<T: Eq> Eq for InternalTuple<T> {}
395
396impl<T: Hash> Hash for InternalTuple<T> {
397 fn hash<H: Hasher>(&self, state: &mut H) {
398 self.inner.hash(state)
399 }
400}
401
402impl<T> IntoIterator for InternalTuple<T> {
403 type Item = Option<T>;
404 type IntoIter = std::vec::IntoIter<Option<T>>;
405
406 fn into_iter(self) -> Self::IntoIter {
407 self.inner.into_iter()
408 }
409}
410
411type InternalTuplesIterator<'a, T> =
412 Box<dyn Iterator<Item = Result<InternalTuple<T>, QueryEvaluationError>> + 'a>;
413
414pub struct SimpleEvaluator<'a, D: QueryableDataset<'a>> {
415 dataset: EvalDataset<'a, D>,
416 base_iri: Option<Arc<Iri<String>>>,
417 now: DateTime,
418 service_handler: Rc<ServiceHandlerRegistry>,
419 custom_functions: Rc<CustomFunctionRegistry>,
420 custom_aggregate_functions: Rc<CustomAggregateFunctionRegistry>,
421 run_stats: bool,
422}
423
424impl<'a, D: QueryableDataset<'a>> SimpleEvaluator<'a, D> {
425 pub fn new(
426 dataset: D,
427 base_iri: Option<Arc<Iri<String>>>,
428 service_handler: Rc<ServiceHandlerRegistry>,
429 custom_functions: Rc<CustomFunctionRegistry>,
430 custom_aggregate_functions: Rc<CustomAggregateFunctionRegistry>,
431 cancellation_token: CancellationToken,
432 dataset_spec: QueryDatasetSpecification,
433 run_stats: bool,
434 ) -> Result<Self, QueryEvaluationError> {
435 Ok(Self {
436 dataset: EvalDataset::new(dataset, dataset_spec, cancellation_token)?,
437 base_iri,
438 now: DateTime::now(),
439 service_handler,
440 custom_functions,
441 custom_aggregate_functions,
442 run_stats,
443 })
444 }
445
446 pub fn evaluate_select(
447 &self,
448 pattern: &GraphPattern,
449 substitutions: impl IntoIterator<Item = (Variable, Term)>,
450 ) -> (
451 Result<QuerySolutionIter<'a>, QueryEvaluationError>,
452 Rc<EvalNodeWithStats>,
453 ) {
454 let mut variables = Vec::new();
455 let (eval, stats) = self.graph_pattern_evaluator(pattern, &mut variables);
456 let eval = match eval {
457 Ok(e) => e,
458 Err(e) => return (Err(e), stats),
459 };
460 let from = match encode_initial_bindings(&self.dataset, &variables, substitutions) {
461 Ok(from) => from,
462 Err(e) => return (Err(e), stats),
463 };
464 (
465 Ok(decode_bindings(
466 self.dataset.clone(),
467 eval(from),
468 Arc::from(variables),
469 )),
470 stats,
471 )
472 }
473
474 pub fn evaluate_ask(
475 &self,
476 pattern: &GraphPattern,
477 substitutions: impl IntoIterator<Item = (Variable, Term)>,
478 ) -> (Result<bool, QueryEvaluationError>, Rc<EvalNodeWithStats>) {
479 let mut variables = Vec::new();
480 let (eval, stats) = self.graph_pattern_evaluator(pattern, &mut variables);
481 let eval = match eval {
482 Ok(e) => e,
483 Err(e) => return (Err(e), stats),
484 };
485 let from = match encode_initial_bindings(&self.dataset, &variables, substitutions) {
486 Ok(from) => from,
487 Err(e) => return (Err(e), stats),
488 };
489 let mut error = None;
492 for solution in eval(from) {
493 if let Err(e) = solution {
494 error.get_or_insert(e);
496 } else {
497 return (Ok(true), stats);
499 }
500 }
501 (
502 if let Some(e) = error {
503 Err(e)
504 } else {
505 Ok(false)
506 },
507 stats,
508 )
509 }
510
511 pub fn evaluate_construct(
512 &self,
513 pattern: &GraphPattern,
514 template: &[TriplePattern],
515 substitutions: impl IntoIterator<Item = (Variable, Term)>,
516 ) -> (
517 Result<QueryTripleIter<'a>, QueryEvaluationError>,
518 Rc<EvalNodeWithStats>,
519 ) {
520 let mut variables = Vec::new();
521 let (eval, stats) = self.graph_pattern_evaluator(pattern, &mut variables);
522 let eval = match eval {
523 Ok(e) => e,
524 Err(e) => return (Err(e), stats),
525 };
526 let mut bnodes = Vec::new();
527 let template = template
528 .iter()
529 .filter_map(|t| {
530 Some(TripleTemplate {
531 subject: TripleTemplateValue::from_term_or_variable(
532 &t.subject,
533 &mut variables,
534 &mut bnodes,
535 )?,
536 predicate: TripleTemplateValue::from_named_node_or_variable(
537 &t.predicate,
538 &mut variables,
539 ),
540 object: TripleTemplateValue::from_term_or_variable(
541 &t.object,
542 &mut variables,
543 &mut bnodes,
544 )?,
545 })
546 })
547 .collect();
548 let from = match encode_initial_bindings(&self.dataset, &variables, substitutions) {
549 Ok(from) => from,
550 Err(e) => return (Err(e), stats),
551 };
552 (
553 Ok(QueryTripleIter::new(ConstructIterator {
554 eval: self.clone(),
555 iter: eval(from),
556 template,
557 buffered_results: Vec::default(),
558 already_emitted_results: FxHashSet::default(),
559 bnodes: Vec::default(),
560 })),
561 stats,
562 )
563 }
564
565 pub fn evaluate_describe(
566 &self,
567 pattern: &GraphPattern,
568 substitutions: impl IntoIterator<Item = (Variable, Term)>,
569 ) -> (
570 Result<QueryTripleIter<'a>, QueryEvaluationError>,
571 Rc<EvalNodeWithStats>,
572 ) {
573 let mut variables = Vec::new();
574 let (eval, stats) = self.graph_pattern_evaluator(pattern, &mut variables);
575 let eval = match eval {
576 Ok(e) => e,
577 Err(e) => return (Err(e), stats),
578 };
579 let from = match encode_initial_bindings(&self.dataset, &variables, substitutions) {
580 Ok(from) => from,
581 Err(e) => return (Err(e), stats),
582 };
583 (
584 Ok(QueryTripleIter::new(DescribeIterator {
585 eval: self.clone(),
586 tuples_to_describe: eval(from),
587 nodes_described: FxHashSet::default(),
588 nodes_to_describe: Vec::default(),
589 quads: Box::new(empty()),
590 })),
591 stats,
592 )
593 }
594
595 pub fn graph_pattern_evaluator(
596 &self,
597 pattern: &GraphPattern,
598 encoded_variables: &mut Vec<Variable>,
599 ) -> (
600 Result<InternalTupleEvaluator<'a, D::InternalTerm>, QueryEvaluationError>,
601 Rc<EvalNodeWithStats>,
602 ) {
603 let mut stat_children = Vec::new();
604 let evaluator =
605 self.build_graph_pattern_evaluator(pattern, encoded_variables, &mut stat_children);
606 let stats = Rc::new(EvalNodeWithStats {
607 label: eval_node_label(pattern),
608 children: stat_children,
609 exec_count: Cell::new(0),
610 exec_duration: Cell::new(self.run_stats.then(DayTimeDuration::default)),
611 });
612 let mut evaluator = match evaluator {
613 Ok(e) => e,
614 Err(e) => return (Err(e), stats),
615 };
616 if self.run_stats {
617 let stats = Rc::clone(&stats);
618 evaluator = Rc::new(move |tuple| {
619 let start = Timer::now();
620 let inner = evaluator(tuple);
621 let duration = start.elapsed();
622 stats.exec_duration.set(
623 stats
624 .exec_duration
625 .get()
626 .and_then(|d| d.checked_add(duration?)),
627 );
628 Box::new(StatsIterator {
629 inner,
630 stats: Rc::clone(&stats),
631 })
632 })
633 }
634 (Ok(evaluator), stats)
635 }
636
637 fn build_graph_pattern_evaluator(
638 &self,
639 pattern: &GraphPattern,
640 encoded_variables: &mut Vec<Variable>,
641 stat_children: &mut Vec<Rc<EvalNodeWithStats>>,
642 ) -> Result<InternalTupleEvaluator<'a, D::InternalTerm>, QueryEvaluationError> {
643 Ok(match pattern {
644 GraphPattern::Values {
645 variables,
646 bindings,
647 } => {
648 let encoding = variables
649 .iter()
650 .map(|v| encode_variable(encoded_variables, v))
651 .collect::<Vec<_>>();
652 let encoded_tuples = bindings
653 .iter()
654 .map(|row| {
655 let mut result = InternalTuple::with_capacity(variables.len());
656 for (key, value) in row.iter().enumerate() {
657 if let Some(term) = value {
658 result.set(
659 encoding[key],
660 match term {
661 GroundTerm::NamedNode(node) => {
662 self.encode_term(node.clone())
663 }
664 GroundTerm::Literal(literal) => {
665 self.encode_term(literal.clone())
666 }
667 #[cfg(feature = "sparql-12")]
668 GroundTerm::Triple(triple) => self.encode_triple(triple),
669 }?,
670 );
671 }
672 }
673 Ok(result)
674 })
675 .collect::<Result<Vec<_>, QueryEvaluationError>>()?;
676 Rc::new(move |from| {
677 Box::new(
678 encoded_tuples
679 .iter()
680 .filter_map(move |t| t.combine_with(&from))
681 .map(Ok)
682 .collect::<Vec<_>>()
683 .into_iter(),
684 )
685 })
686 }
687 GraphPattern::QuadPattern {
688 subject,
689 predicate,
690 object,
691 graph_name,
692 } => {
693 let subject_selector = TupleSelector::from_ground_term_pattern(
694 subject,
695 encoded_variables,
696 &self.dataset,
697 )?;
698 let predicate_selector = TupleSelector::from_named_node_pattern(
699 predicate,
700 encoded_variables,
701 &self.dataset,
702 )?;
703 let object_selector = TupleSelector::from_ground_term_pattern(
704 object,
705 encoded_variables,
706 &self.dataset,
707 )?;
708 let graph_name_selector = if let Some(graph_name) = graph_name.as_ref() {
709 Some(TupleSelector::from_named_node_pattern(
710 graph_name,
711 encoded_variables,
712 &self.dataset,
713 )?)
714 } else {
715 None
716 };
717 let dataset = self.dataset.clone();
718 Rc::new(move |from| {
719 let input_subject = match subject_selector.get_pattern_value(
720 &from,
721 #[cfg(feature = "sparql-12")]
722 &dataset,
723 ) {
724 Ok(value) => value,
725 Err(e) => return Box::new(once(Err(e))),
726 };
727 let input_predicate = match predicate_selector.get_pattern_value(
728 &from,
729 #[cfg(feature = "sparql-12")]
730 &dataset,
731 ) {
732 Ok(value) => value,
733 Err(e) => return Box::new(once(Err(e))),
734 };
735 let input_object = match object_selector.get_pattern_value(
736 &from,
737 #[cfg(feature = "sparql-12")]
738 &dataset,
739 ) {
740 Ok(value) => value,
741 Err(e) => return Box::new(once(Err(e))),
742 };
743 let input_graph_name = if let Some(graph_name_selector) = &graph_name_selector {
744 match graph_name_selector.get_pattern_value(
745 &from,
746 #[cfg(feature = "sparql-12")]
747 &dataset,
748 ) {
749 Ok(value) => value,
750 Err(e) => return Box::new(once(Err(e))),
751 }
752 .map(Some)
753 } else {
754 Some(None) };
756 let iter = dataset.internal_quads_for_pattern(
757 input_subject.as_ref(),
758 input_predicate.as_ref(),
759 input_object.as_ref(),
760 input_graph_name.as_ref().map(|g| g.as_ref()),
761 );
762 let subject_selector = subject_selector.clone();
763 let predicate_selector = predicate_selector.clone();
764 let object_selector = object_selector.clone();
765 let graph_name_selector = graph_name_selector.clone();
766 #[cfg(feature = "sparql-12")]
767 let dataset = dataset.clone();
768 Box::new(
769 iter.map(move |quad| {
770 let quad = quad?;
771 let mut new_tuple = from.clone();
772 if !put_pattern_value::<D>(
773 &subject_selector,
774 quad.subject,
775 &mut new_tuple,
776 #[cfg(feature = "sparql-12")]
777 &dataset,
778 )? {
779 return Ok(None);
780 }
781 if !put_pattern_value::<D>(
782 &predicate_selector,
783 quad.predicate,
784 &mut new_tuple,
785 #[cfg(feature = "sparql-12")]
786 &dataset,
787 )? {
788 return Ok(None);
789 }
790 if !put_pattern_value::<D>(
791 &object_selector,
792 quad.object,
793 &mut new_tuple,
794 #[cfg(feature = "sparql-12")]
795 &dataset,
796 )? {
797 return Ok(None);
798 }
799 if let Some(graph_name_selector) = &graph_name_selector {
800 let Some(quad_graph_name) = quad.graph_name else {
801 return Err(QueryEvaluationError::UnexpectedDefaultGraph);
802 };
803 if !put_pattern_value::<D>(
804 graph_name_selector,
805 quad_graph_name,
806 &mut new_tuple,
807 #[cfg(feature = "sparql-12")]
808 &dataset,
809 )? {
810 return Ok(None);
811 }
812 }
813 Ok(Some(new_tuple))
814 })
815 .filter_map(Result::transpose),
816 )
817 })
818 }
819 GraphPattern::Path {
820 subject,
821 path,
822 object,
823 graph_name,
824 } => {
825 let subject_selector = TupleSelector::from_ground_term_pattern(
826 subject,
827 encoded_variables,
828 &self.dataset,
829 )?;
830 let path = self.encode_property_path(path)?;
831 let object_selector = TupleSelector::from_ground_term_pattern(
832 object,
833 encoded_variables,
834 &self.dataset,
835 )?;
836 let graph_name_selector = if let Some(graph_name) = graph_name.as_ref() {
837 Some(TupleSelector::from_named_node_pattern(
838 graph_name,
839 encoded_variables,
840 &self.dataset,
841 )?)
842 } else {
843 None
844 };
845 let dataset = self.dataset.clone();
846 Rc::new(move |from| {
847 let input_subject = match subject_selector.get_pattern_value(
848 &from,
849 #[cfg(feature = "sparql-12")]
850 &dataset,
851 ) {
852 Ok(value) => value,
853 Err(e) => return Box::new(once(Err(e))),
854 };
855 let path_eval = PathEvaluator {
856 dataset: dataset.clone(),
857 };
858 let input_object = match object_selector.get_pattern_value(
859 &from,
860 #[cfg(feature = "sparql-12")]
861 &dataset,
862 ) {
863 Ok(value) => value,
864 Err(e) => return Box::new(once(Err(e))),
865 };
866 let input_graph_name = if let Some(graph_name_selector) = &graph_name_selector {
867 match graph_name_selector.get_pattern_value(
868 &from,
869 #[cfg(feature = "sparql-12")]
870 &dataset,
871 ) {
872 Ok(value) => value,
873 Err(e) => return Box::new(once(Err(e))),
874 }
875 .map(Some)
876 } else {
877 Some(None) };
879 match (input_subject, input_object, input_graph_name) {
880 (Some(input_subject), Some(input_object), Some(input_graph_name)) => {
881 match path_eval.eval_closed_in_graph(
882 &path,
883 &input_subject,
884 &input_object,
885 input_graph_name.as_ref(),
886 ) {
887 Ok(true) => Box::new(once(Ok(from))),
888 Ok(false) => Box::new(empty()),
889 Err(e) => Box::new(once(Err(e))),
890 }
891 }
892 (Some(input_subject), None, Some(input_graph_name)) => {
893 let object_selector = object_selector.clone();
894 #[cfg(feature = "sparql-12")]
895 let dataset = dataset.clone();
896 Box::new(
897 path_eval
898 .eval_from_in_graph(
899 &path,
900 &input_subject,
901 input_graph_name.as_ref(),
902 )
903 .map(move |o| {
904 let o = o?;
905 let mut new_tuple = from.clone();
906 if !put_pattern_value::<D>(
907 &object_selector,
908 o,
909 &mut new_tuple,
910 #[cfg(feature = "sparql-12")]
911 &dataset,
912 )? {
913 return Ok(None);
914 }
915 Ok(Some(new_tuple))
916 })
917 .filter_map(Result::transpose),
918 )
919 }
920 (None, Some(input_object), Some(input_graph_name)) => {
921 let subject_selector = subject_selector.clone();
922 #[cfg(feature = "sparql-12")]
923 let dataset = dataset.clone();
924 Box::new(
925 path_eval
926 .eval_to_in_graph(
927 &path,
928 &input_object,
929 input_graph_name.as_ref(),
930 )
931 .map(move |s| {
932 let s = s?;
933 let mut new_tuple = from.clone();
934 if !put_pattern_value::<D>(
935 &subject_selector,
936 s,
937 &mut new_tuple,
938 #[cfg(feature = "sparql-12")]
939 &dataset,
940 )? {
941 return Ok(None);
942 }
943 Ok(Some(new_tuple))
944 })
945 .filter_map(Result::transpose),
946 )
947 }
948 (None, None, Some(input_graph_name)) => {
949 let subject_selector = subject_selector.clone();
950 let object_selector = object_selector.clone();
951 #[cfg(feature = "sparql-12")]
952 let dataset = dataset.clone();
953 Box::new(
954 path_eval
955 .eval_open_in_graph(&path, input_graph_name.as_ref())
956 .map(move |t| {
957 let (s, o) = t?;
958 let mut new_tuple = from.clone();
959 if !put_pattern_value::<D>(
960 &subject_selector,
961 s,
962 &mut new_tuple,
963 #[cfg(feature = "sparql-12")]
964 &dataset,
965 )? {
966 return Ok(None);
967 }
968 if !put_pattern_value::<D>(
969 &object_selector,
970 o,
971 &mut new_tuple,
972 #[cfg(feature = "sparql-12")]
973 &dataset,
974 )? {
975 return Ok(None);
976 }
977 Ok(Some(new_tuple))
978 })
979 .filter_map(Result::transpose),
980 )
981 }
982 (Some(input_subject), Some(input_object), None) => {
983 let graph_name_selector = graph_name_selector.clone();
984 #[cfg(feature = "sparql-12")]
985 let dataset = dataset.clone();
986 Box::new(
987 path_eval
988 .eval_closed_in_unknown_graph(
989 &path,
990 &input_subject,
991 &input_object,
992 )
993 .map(move |g| {
994 let g = g?;
995 let mut new_tuple = from.clone();
996 if let Some(graph_name_selector) = &graph_name_selector {
997 let Some(g) = g else {
998 return Err(
999 QueryEvaluationError::UnexpectedDefaultGraph,
1000 );
1001 };
1002 if !put_pattern_value::<D>(
1003 graph_name_selector,
1004 g,
1005 &mut new_tuple,
1006 #[cfg(feature = "sparql-12")]
1007 &dataset,
1008 )? {
1009 return Ok(None);
1010 }
1011 }
1012 Ok(Some(new_tuple))
1013 })
1014 .filter_map(Result::transpose),
1015 )
1016 }
1017 (Some(input_subject), None, None) => {
1018 let object_selector = object_selector.clone();
1019 let graph_name_selector = graph_name_selector.clone();
1020 #[cfg(feature = "sparql-12")]
1021 let dataset = dataset.clone();
1022 Box::new(
1023 path_eval
1024 .eval_from_in_unknown_graph(&path, &input_subject)
1025 .map(move |t| {
1026 let (o, g) = t?;
1027 let mut new_tuple = from.clone();
1028 if !put_pattern_value::<D>(
1029 &object_selector,
1030 o,
1031 &mut new_tuple,
1032 #[cfg(feature = "sparql-12")]
1033 &dataset,
1034 )? {
1035 return Ok(None);
1036 }
1037 if let Some(graph_name_selector) = &graph_name_selector {
1038 let Some(g) = g else {
1039 return Err(
1040 QueryEvaluationError::UnexpectedDefaultGraph,
1041 );
1042 };
1043 if !put_pattern_value::<D>(
1044 graph_name_selector,
1045 g,
1046 &mut new_tuple,
1047 #[cfg(feature = "sparql-12")]
1048 &dataset,
1049 )? {
1050 return Ok(None);
1051 }
1052 }
1053 Ok(Some(new_tuple))
1054 })
1055 .filter_map(Result::transpose),
1056 )
1057 }
1058 (None, Some(input_object), None) => {
1059 let subject_selector = subject_selector.clone();
1060 let graph_name_selector = graph_name_selector.clone();
1061 #[cfg(feature = "sparql-12")]
1062 let dataset = dataset.clone();
1063 Box::new(
1064 path_eval
1065 .eval_to_in_unknown_graph(&path, &input_object)
1066 .map(move |t| {
1067 let (s, g) = t?;
1068 let mut new_tuple = from.clone();
1069 if !put_pattern_value::<D>(
1070 &subject_selector,
1071 s,
1072 &mut new_tuple,
1073 #[cfg(feature = "sparql-12")]
1074 &dataset,
1075 )? {
1076 return Ok(None);
1077 }
1078 if let Some(graph_name_selector) = &graph_name_selector {
1079 let Some(g) = g else {
1080 return Err(
1081 QueryEvaluationError::UnexpectedDefaultGraph,
1082 );
1083 };
1084 if !put_pattern_value::<D>(
1085 graph_name_selector,
1086 g,
1087 &mut new_tuple,
1088 #[cfg(feature = "sparql-12")]
1089 &dataset,
1090 )? {
1091 return Ok(None);
1092 }
1093 }
1094 Ok(Some(new_tuple))
1095 })
1096 .filter_map(Result::transpose),
1097 )
1098 }
1099 (None, None, None) => {
1100 let subject_selector = subject_selector.clone();
1101 let object_selector = object_selector.clone();
1102 let graph_name_selector = graph_name_selector.clone();
1103 #[cfg(feature = "sparql-12")]
1104 let dataset = dataset.clone();
1105 Box::new(
1106 path_eval
1107 .eval_open_in_unknown_graph(&path)
1108 .map(move |t| {
1109 let (s, o, g) = t?;
1110 let mut new_tuple = from.clone();
1111 if !put_pattern_value::<D>(
1112 &subject_selector,
1113 s,
1114 &mut new_tuple,
1115 #[cfg(feature = "sparql-12")]
1116 &dataset,
1117 )? {
1118 return Ok(None);
1119 }
1120 if !put_pattern_value::<D>(
1121 &object_selector,
1122 o,
1123 &mut new_tuple,
1124 #[cfg(feature = "sparql-12")]
1125 &dataset,
1126 )? {
1127 return Ok(None);
1128 }
1129 if let Some(graph_name_selector) = &graph_name_selector {
1130 let Some(g) = g else {
1131 return Err(
1132 QueryEvaluationError::UnexpectedDefaultGraph,
1133 );
1134 };
1135 if !put_pattern_value::<D>(
1136 graph_name_selector,
1137 g,
1138 &mut new_tuple,
1139 #[cfg(feature = "sparql-12")]
1140 &dataset,
1141 )? {
1142 return Ok(None);
1143 }
1144 }
1145 Ok(Some(new_tuple))
1146 })
1147 .filter_map(Result::transpose),
1148 )
1149 }
1150 }
1151 })
1152 }
1153 GraphPattern::Graph { graph_name } => {
1154 let graph_name_selector = TupleSelector::from_named_node_pattern(
1155 graph_name,
1156 encoded_variables,
1157 &self.dataset,
1158 )?;
1159 let dataset = self.dataset.clone();
1160 Rc::new(move |from| {
1161 let input_graph_name = match graph_name_selector.get_pattern_value(
1162 &from,
1163 #[cfg(feature = "sparql-12")]
1164 &dataset,
1165 ) {
1166 Ok(value) => value,
1167 Err(e) => return Box::new(once(Err(e))),
1168 };
1169 if let Some(input_graph_name) = input_graph_name {
1170 match dataset.contains_internal_graph_name(&input_graph_name) {
1171 Ok(true) => Box::new(once(Ok(from))),
1172 Ok(false) => Box::new(empty()),
1173 Err(e) => Box::new(once(Err(e))),
1174 }
1175 } else {
1176 let graph_name_selector = graph_name_selector.clone();
1177 #[cfg(feature = "sparql-12")]
1178 let dataset = dataset.clone();
1179 Box::new(
1180 dataset
1181 .internal_named_graphs()
1182 .map(move |graph_name| {
1183 let graph_name = graph_name?;
1184 let mut new_tuple = from.clone();
1185 if !put_pattern_value::<D>(
1186 &graph_name_selector,
1187 graph_name,
1188 &mut new_tuple,
1189 #[cfg(feature = "sparql-12")]
1190 &dataset,
1191 )? {
1192 return Ok(None);
1193 }
1194 Ok(Some(new_tuple))
1195 })
1196 .filter_map(Result::transpose),
1197 )
1198 }
1199 })
1200 }
1201 GraphPattern::Join {
1202 left,
1203 right,
1204 algorithm,
1205 } => {
1206 let (left, left_stats) = self.graph_pattern_evaluator(left, encoded_variables);
1207 stat_children.push(left_stats);
1208 let (right, right_stats) = self.graph_pattern_evaluator(right, encoded_variables);
1209 stat_children.push(right_stats);
1210 let left = left?;
1211 let right = right?;
1212
1213 match algorithm {
1214 JoinAlgorithm::HashBuildLeftProbeRight { keys } => {
1215 let build = left;
1216 let probe = right;
1217 if keys.is_empty() {
1218 Rc::new(move |from| {
1220 let mut errors = Vec::default();
1221 let built_values = build(from.clone())
1222 .filter_map(|result| match result {
1223 Ok(result) => Some(result),
1224 Err(error) => {
1225 errors.push(Err(error));
1226 None
1227 }
1228 })
1229 .collect::<Vec<_>>();
1230 if built_values.is_empty() && errors.is_empty() {
1231 return Box::new(empty());
1233 }
1234 let mut probe_iter = probe(from).peekable();
1235 if probe_iter.peek().is_none() {
1236 return Box::new(empty());
1238 }
1239 Box::new(CartesianProductJoinIterator {
1240 probe_iter,
1241 built: built_values,
1242 buffered_results: errors,
1243 })
1244 })
1245 } else {
1246 let keys = keys
1248 .iter()
1249 .map(|v| encode_variable(encoded_variables, v))
1250 .collect::<Vec<_>>();
1251 Rc::new(move |from| {
1252 let mut errors = Vec::default();
1253 let mut built_values = InternalTupleSet::new(keys.clone());
1254 built_values.extend(build(from.clone()).filter_map(|result| {
1255 match result {
1256 Ok(result) => Some(result),
1257 Err(error) => {
1258 errors.push(Err(error));
1259 None
1260 }
1261 }
1262 }));
1263 if built_values.is_empty() && errors.is_empty() {
1264 return Box::new(empty());
1266 }
1267 let mut probe_iter = probe(from).peekable();
1268 if probe_iter.peek().is_none() {
1269 return Box::new(empty());
1271 }
1272 Box::new(HashJoinIterator {
1273 probe_iter,
1274 built: built_values,
1275 buffered_results: errors,
1276 })
1277 })
1278 }
1279 }
1280 }
1281 }
1282 #[cfg(feature = "sep-0006")]
1283 GraphPattern::Lateral { left, right } => {
1284 let (left, left_stats) = self.graph_pattern_evaluator(left, encoded_variables);
1285 stat_children.push(left_stats);
1286 let left = left?;
1287
1288 if let GraphPattern::LeftJoin {
1289 left: nested_left,
1290 right: nested_right,
1291 expression,
1292 ..
1293 } = right.as_ref()
1294 {
1295 if nested_left.is_empty_singleton() {
1296 let right =
1298 GraphPattern::filter(nested_right.as_ref().clone(), expression.clone());
1299 let (right, right_stats) =
1300 self.graph_pattern_evaluator(&right, encoded_variables);
1301 stat_children.push(right_stats);
1302 let right = right?;
1303 return Ok(Rc::new(move |from| {
1304 Box::new(ForLoopLeftJoinIterator {
1305 right_evaluator: Rc::clone(&right),
1306 left_iter: left(from),
1307 current_right: Box::new(empty()),
1308 left_tuple_to_yield: None,
1309 })
1310 }));
1311 }
1312 }
1313 let (right, right_stats) = self.graph_pattern_evaluator(right, encoded_variables);
1314 stat_children.push(right_stats);
1315 let right = right?;
1316 Rc::new(move |from| {
1317 let right = Rc::clone(&right);
1318 Box::new(left(from).flat_map(move |t| match t {
1319 Ok(t) => right(t),
1320 Err(e) => Box::new(once(Err(e))),
1321 }))
1322 })
1323 }
1324 GraphPattern::Minus {
1325 left,
1326 right,
1327 algorithm,
1328 } => {
1329 let (left, left_stats) = self.graph_pattern_evaluator(left, encoded_variables);
1330 stat_children.push(left_stats);
1331 let (right, right_stats) = self.graph_pattern_evaluator(right, encoded_variables);
1332 stat_children.push(right_stats);
1333 let left = left?;
1334 let right = right?;
1335
1336 match algorithm {
1337 MinusAlgorithm::HashBuildRightProbeLeft { keys } => {
1338 if keys.is_empty() {
1339 Rc::new(move |from| {
1340 let right: Vec<_> =
1341 right(from.clone()).filter_map(Result::ok).collect();
1342 if right.is_empty() {
1343 return left(from);
1344 }
1345 Box::new(left(from).filter(move |left_tuple| {
1346 if let Ok(left_tuple) = left_tuple {
1347 !right.iter().any(|right_tuple| {
1348 are_compatible_and_not_disjointed(
1349 left_tuple,
1350 right_tuple,
1351 )
1352 })
1353 } else {
1354 true
1355 }
1356 }))
1357 })
1358 } else {
1359 let keys = keys
1360 .iter()
1361 .map(|v| encode_variable(encoded_variables, v))
1362 .collect::<Vec<_>>();
1363 Rc::new(move |from| {
1364 let mut right_values = InternalTupleSet::new(keys.clone());
1365 right_values.extend(right(from.clone()).filter_map(Result::ok));
1366 if right_values.is_empty() {
1367 return left(from);
1368 }
1369 Box::new(left(from).filter(move |left_tuple| {
1370 if let Ok(left_tuple) = left_tuple {
1371 !right_values.get(left_tuple).iter().any(|right_tuple| {
1372 are_compatible_and_not_disjointed(
1373 left_tuple,
1374 right_tuple,
1375 )
1376 })
1377 } else {
1378 true
1379 }
1380 }))
1381 })
1382 }
1383 }
1384 }
1385 }
1386 GraphPattern::LeftJoin {
1387 left,
1388 right,
1389 expression,
1390 algorithm,
1391 } => {
1392 let (left, left_stats) = self.graph_pattern_evaluator(left, encoded_variables);
1393 stat_children.push(left_stats);
1394 let (right, right_stats) = self.graph_pattern_evaluator(right, encoded_variables);
1395 stat_children.push(right_stats);
1396 let left = left?;
1397 let right = right?;
1398 let expression = self.effective_boolean_value_expression_evaluator(
1399 expression,
1400 encoded_variables,
1401 stat_children,
1402 )?;
1403
1404 match algorithm {
1405 LeftJoinAlgorithm::HashBuildRightProbeLeft { keys } => {
1406 let keys = keys
1408 .iter()
1409 .map(|v| encode_variable(encoded_variables, v))
1410 .collect::<Vec<_>>();
1411 Rc::new(move |from| {
1412 let mut errors = Vec::default();
1413 let mut right_values = InternalTupleSet::new(keys.clone());
1414 right_values.extend(right(from.clone()).filter_map(
1415 |result| match result {
1416 Ok(result) => Some(result),
1417 Err(error) => {
1418 errors.push(Err(error));
1419 None
1420 }
1421 },
1422 ));
1423 if right_values.is_empty() && errors.is_empty() {
1424 return left(from);
1425 }
1426 Box::new(HashLeftJoinIterator {
1427 left_iter: left(from),
1428 right: right_values,
1429 buffered_results: errors,
1430 expression: Rc::clone(&expression),
1431 })
1432 })
1433 }
1434 }
1435 }
1436 GraphPattern::Filter { inner, expression } => {
1437 let (child, child_stats) = self.graph_pattern_evaluator(inner, encoded_variables);
1438 stat_children.push(child_stats);
1439 let child = child?;
1440 let expression = self.effective_boolean_value_expression_evaluator(
1441 expression,
1442 encoded_variables,
1443 stat_children,
1444 )?;
1445 Rc::new(move |from| {
1446 let expression = Rc::clone(&expression);
1447 Box::new(child(from).filter(move |tuple| match tuple {
1448 Ok(tuple) => expression(tuple).unwrap_or(false),
1449 Err(_) => true,
1450 }))
1451 })
1452 }
1453 GraphPattern::Union { inner } => {
1454 let children = inner
1455 .iter()
1456 .map(|child| {
1457 let (child, child_stats) =
1458 self.graph_pattern_evaluator(child, encoded_variables);
1459 stat_children.push(child_stats);
1460 child
1461 })
1462 .collect::<Result<Vec<_>, _>>()?;
1463
1464 Rc::new(move |from| {
1465 Box::new(UnionIterator {
1466 plans: children.clone(),
1467 input: from,
1468 current_iterator: Box::new(empty()),
1469 current_plan: 0,
1470 })
1471 })
1472 }
1473 GraphPattern::Extend {
1474 inner,
1475 variable,
1476 expression,
1477 } => {
1478 let (child, child_stats) = self.graph_pattern_evaluator(inner, encoded_variables);
1479 stat_children.push(child_stats);
1480 let child = child?;
1481
1482 let position = encode_variable(encoded_variables, variable);
1483 if let Some(expression) = self.internal_expression_evaluator(
1484 expression,
1485 encoded_variables,
1486 stat_children,
1487 )? {
1488 return Ok(Rc::new(move |from| {
1489 let expression = Rc::clone(&expression);
1490 Box::new(child(from).map(move |tuple| {
1491 let mut tuple = tuple?;
1492 if let Some(value) = expression(&tuple) {
1493 tuple.set(position, value);
1494 }
1495 Ok(tuple)
1496 }))
1497 }));
1498 }
1499
1500 let expression =
1501 self.expression_evaluator(expression, encoded_variables, stat_children)?;
1502 let dataset = self.dataset.clone();
1503 Rc::new(move |from| {
1504 let expression = Rc::clone(&expression);
1505 let dataset = dataset.clone();
1506 Box::new(child(from).map(move |tuple| {
1507 let mut tuple = tuple?;
1508 if let Some(value) = expression(&tuple) {
1509 tuple.set(position, dataset.internalize_expression_term(value)?);
1510 }
1511 Ok(tuple)
1512 }))
1513 })
1514 }
1515 GraphPattern::OrderBy { inner, expression } => {
1516 let (child, child_stats) = self.graph_pattern_evaluator(inner, encoded_variables);
1517 stat_children.push(child_stats);
1518 let child = child?;
1519 let by = expression
1520 .iter()
1521 .map(|comp| {
1522 Ok(match comp {
1523 OrderExpression::Asc(expression) => {
1524 ComparatorFunction::Asc(self.expression_evaluator(
1525 expression,
1526 encoded_variables,
1527 stat_children,
1528 )?)
1529 }
1530 OrderExpression::Desc(expression) => {
1531 ComparatorFunction::Desc(self.expression_evaluator(
1532 expression,
1533 encoded_variables,
1534 stat_children,
1535 )?)
1536 }
1537 })
1538 })
1539 .collect::<Result<Vec<_>, QueryEvaluationError>>()?;
1540 Rc::new(move |from| {
1541 let mut errors = Vec::default();
1542 let mut values = child(from)
1543 .filter_map(|result| match result {
1544 Ok(result) => Some(result),
1545 Err(error) => {
1546 errors.push(Err(error));
1547 None
1548 }
1549 })
1550 .collect::<Vec<_>>();
1551 values.sort_unstable_by(|a, b| {
1552 for comp in &by {
1553 match comp {
1554 ComparatorFunction::Asc(expression) => {
1555 match cmp_terms(expression(a).as_ref(), expression(b).as_ref())
1556 {
1557 Ordering::Greater => return Ordering::Greater,
1558 Ordering::Less => return Ordering::Less,
1559 Ordering::Equal => (),
1560 }
1561 }
1562 ComparatorFunction::Desc(expression) => {
1563 match cmp_terms(expression(a).as_ref(), expression(b).as_ref())
1564 {
1565 Ordering::Greater => return Ordering::Less,
1566 Ordering::Less => return Ordering::Greater,
1567 Ordering::Equal => (),
1568 }
1569 }
1570 }
1571 }
1572 Ordering::Equal
1573 });
1574 Box::new(errors.into_iter().chain(values.into_iter().map(Ok)))
1575 })
1576 }
1577 GraphPattern::Distinct { inner } => {
1578 let (child, child_stats) = self.graph_pattern_evaluator(inner, encoded_variables);
1579 stat_children.push(child_stats);
1580 let child = child?;
1581 Rc::new(move |from| Box::new(hash_deduplicate(child(from))))
1582 }
1583 GraphPattern::Reduced { inner } => {
1584 let (child, child_stats) = self.graph_pattern_evaluator(inner, encoded_variables);
1585 stat_children.push(child_stats);
1586 let child = child?;
1587 Rc::new(move |from| {
1588 Box::new(ConsecutiveDeduplication {
1589 inner: child(from),
1590 current: None,
1591 })
1592 })
1593 }
1594 GraphPattern::Slice {
1595 inner,
1596 start,
1597 length,
1598 } => {
1599 let (child, child_stats) = self.graph_pattern_evaluator(inner, encoded_variables);
1600 stat_children.push(child_stats);
1601 let mut child = child?;
1602 #[expect(clippy::shadow_same)]
1603 let start = *start;
1604 if start > 0 {
1605 child = Rc::new(move |from| Box::new(child(from).skip(start)));
1606 }
1607 if let Some(length) = *length {
1608 child = Rc::new(move |from| Box::new(child(from).take(length)));
1609 }
1610 child
1611 }
1612 GraphPattern::Project { inner, variables } => {
1613 let mut inner_encoded_variables = variables.clone();
1614 let (child, child_stats) =
1615 self.graph_pattern_evaluator(inner, &mut inner_encoded_variables);
1616 stat_children.push(child_stats);
1617 let child = child?;
1618 let mapping = variables
1619 .iter()
1620 .enumerate()
1621 .map(|(new_variable, variable)| {
1622 (new_variable, encode_variable(encoded_variables, variable))
1623 })
1624 .collect::<Rc<[(usize, usize)]>>();
1625 Rc::new(move |from| {
1626 let mapping = Rc::clone(&mapping);
1627 let mut input_tuple = InternalTuple::with_capacity(mapping.len());
1628 for (input_key, output_key) in &*mapping {
1629 if let Some(value) = from.get(*output_key) {
1630 input_tuple.set(*input_key, value.clone());
1631 }
1632 }
1633 Box::new(child(input_tuple).filter_map(move |tuple| {
1634 match tuple {
1635 Ok(tuple) => {
1636 let mut output_tuple = from.clone();
1637 for (input_key, output_key) in &*mapping {
1638 if let Some(value) = tuple.get(*input_key) {
1639 if let Some(existing_value) = output_tuple.get(*output_key)
1640 {
1641 if existing_value != value {
1642 return None; }
1644 } else {
1645 output_tuple.set(*output_key, value.clone());
1646 }
1647 }
1648 }
1649 Some(Ok(output_tuple))
1650 }
1651 Err(e) => Some(Err(e)),
1652 }
1653 }))
1654 })
1655 }
1656 GraphPattern::Group {
1657 inner,
1658 aggregates,
1659 variables,
1660 } => {
1661 let (child, child_stats) = self.graph_pattern_evaluator(inner, encoded_variables);
1662 stat_children.push(child_stats);
1663 let child = child?;
1664 let key_variables = variables
1665 .iter()
1666 .map(|k| encode_variable(encoded_variables, k))
1667 .collect::<Rc<[_]>>();
1668 let accumulator_builders = aggregates
1669 .iter()
1670 .map(|(_, aggregate)| {
1671 self.accumulator_builder(aggregate, encoded_variables, stat_children)
1672 })
1673 .collect::<Result<Vec<_>, _>>()?;
1674 let accumulator_variables = aggregates
1675 .iter()
1676 .map(|(variable, _)| encode_variable(encoded_variables, variable))
1677 .collect::<Vec<_>>();
1678 let dataset = self.dataset.clone();
1679 Rc::new(move |from| {
1680 let tuple_size = from.capacity();
1681 let key_variables = Rc::clone(&key_variables);
1682 let mut errors = Vec::default();
1683 let mut accumulators_for_group = FxHashMap::<
1684 Vec<Option<D::InternalTerm>>,
1685 Vec<AccumulatorWrapper<'_, D::InternalTerm>>,
1686 >::default();
1687 if key_variables.is_empty() {
1688 accumulators_for_group.insert(
1690 Vec::new(),
1691 accumulator_builders.iter().map(|c| c()).collect::<Vec<_>>(),
1692 );
1693 }
1694 child(from)
1695 .filter_map(|result| match result {
1696 Ok(result) => Some(result),
1697 Err(error) => {
1698 errors.push(error);
1699 None
1700 }
1701 })
1702 .for_each(|tuple| {
1703 let key = key_variables
1705 .iter()
1706 .map(|v| tuple.get(*v).cloned())
1707 .collect();
1708
1709 let key_accumulators =
1710 accumulators_for_group.entry(key).or_insert_with(|| {
1711 accumulator_builders.iter().map(|c| c()).collect::<Vec<_>>()
1712 });
1713 for accumulator in key_accumulators {
1714 accumulator.accumulate(&tuple);
1715 }
1716 });
1717 let accumulator_variables = accumulator_variables.clone();
1718 let dataset = dataset.clone();
1719 Box::new(
1720 errors
1721 .into_iter()
1722 .map(Err)
1723 .chain(accumulators_for_group.into_iter().map(
1724 move |(key, accumulators)| {
1725 let mut result = InternalTuple::with_capacity(tuple_size);
1726 for (variable, value) in key_variables.iter().zip(key) {
1727 if let Some(value) = value {
1728 result.set(*variable, value);
1729 }
1730 }
1731 for (accumulator, variable) in
1732 accumulators.into_iter().zip(&accumulator_variables)
1733 {
1734 if let Some(value) = accumulator.finish() {
1735 result.set(
1736 *variable,
1737 dataset.internalize_expression_term(value)?,
1738 );
1739 }
1740 }
1741 Ok(result)
1742 },
1743 )),
1744 )
1745 })
1746 }
1747 GraphPattern::Service {
1748 name,
1749 inner,
1750 silent,
1751 } => {
1752 #[expect(clippy::shadow_same)]
1753 let silent = *silent;
1754 let service_name =
1755 TupleSelector::from_named_node_pattern(name, encoded_variables, &self.dataset)?;
1756 inner.lookup_used_variables(&mut |v| {
1757 encode_variable(encoded_variables, v);
1758 }); let graph_pattern = spargebra::algebra::GraphPattern::from(inner.as_ref());
1760 let variables = Rc::from(encoded_variables.as_slice());
1761 let eval = self.clone();
1762 Rc::new(move |from| {
1763 match eval.evaluate_service(
1764 &service_name,
1765 &graph_pattern,
1766 Rc::clone(&variables),
1767 &from,
1768 ) {
1769 Ok(result) => Box::new(result.filter_map(move |binding| {
1770 binding
1771 .map(|binding| binding.combine_with(&from))
1772 .transpose()
1773 })),
1774 Err(e) => {
1775 if silent {
1776 Box::new(once(Ok(from)))
1777 } else {
1778 Box::new(once(Err(e)))
1779 }
1780 }
1781 }
1782 })
1783 }
1784 })
1785 }
1786
1787 fn evaluate_service(
1788 &self,
1789 service_name: &TupleSelector<D::InternalTerm>,
1790 graph_pattern: &spargebra::algebra::GraphPattern,
1791 variables: Rc<[Variable]>,
1792 from: &InternalTuple<D::InternalTerm>,
1793 ) -> Result<InternalTuplesIterator<'a, D::InternalTerm>, QueryEvaluationError> {
1794 let service_name = service_name
1795 .get_pattern_value(
1796 from,
1797 #[cfg(feature = "sparql-12")]
1798 &self.dataset,
1799 )?
1800 .ok_or(QueryEvaluationError::UnboundService)?;
1801 let service_name = match self.dataset.externalize_term(service_name)? {
1802 Term::NamedNode(service_name) => service_name,
1803 term => return Err(QueryEvaluationError::InvalidServiceName(term)),
1804 };
1805 let iter =
1806 self.service_handler
1807 .handle(&service_name, graph_pattern, self.base_iri.as_deref())?;
1808 Ok(encode_bindings(self.dataset.clone(), variables, iter))
1809 }
1810
1811 fn accumulator_builder(
1812 &self,
1813 expression: &AggregateExpression,
1814 encoded_variables: &mut Vec<Variable>,
1815 stat_children: &mut Vec<Rc<EvalNodeWithStats>>,
1816 ) -> Result<Box<dyn Fn() -> AccumulatorWrapper<'a, D::InternalTerm> + 'a>, QueryEvaluationError>
1817 {
1818 Ok(match expression {
1819 AggregateExpression::CountSolutions { distinct } => {
1820 if *distinct {
1821 Box::new(move || AccumulatorWrapper::CountDistinctTuple {
1822 count: 0,
1823 seen: FxHashSet::default(),
1824 })
1825 } else {
1826 Box::new(move || AccumulatorWrapper::CountTuple { count: 0 })
1827 }
1828 }
1829 AggregateExpression::FunctionCall {
1830 name,
1831 distinct,
1832 expr,
1833 } => match name {
1834 AggregateFunction::Count => {
1835 if let Some(evaluator) =
1836 self.internal_expression_evaluator(expr, encoded_variables, stat_children)?
1837 {
1838 return Ok(if *distinct {
1839 Box::new(move || AccumulatorWrapper::CountDistinctInternal {
1840 evaluator: Rc::clone(&evaluator),
1841 seen: FxHashSet::default(),
1842 count: 0,
1843 })
1844 } else {
1845 Box::new(move || AccumulatorWrapper::CountInternal {
1846 evaluator: Rc::clone(&evaluator),
1847 count: 0,
1848 })
1849 });
1850 }
1851 let evaluator =
1852 self.expression_evaluator(expr, encoded_variables, stat_children)?;
1853 if *distinct {
1854 Box::new(move || AccumulatorWrapper::DistinctExpression {
1855 evaluator: Rc::clone(&evaluator),
1856 seen: FxHashSet::default(),
1857 accumulator: Some(Box::new(CountAccumulator::default())),
1858 })
1859 } else {
1860 Box::new(move || AccumulatorWrapper::Expression {
1861 evaluator: Rc::clone(&evaluator),
1862 accumulator: Some(Box::new(CountAccumulator::default())),
1863 })
1864 }
1865 }
1866 AggregateFunction::Sum => {
1867 let evaluator =
1868 self.expression_evaluator(expr, encoded_variables, stat_children)?;
1869 if *distinct {
1870 Box::new(move || AccumulatorWrapper::DistinctExpression {
1871 evaluator: Rc::clone(&evaluator),
1872 seen: FxHashSet::default(),
1873 accumulator: Some(Box::new(SumAccumulator::default())),
1874 })
1875 } else {
1876 Box::new(move || AccumulatorWrapper::Expression {
1877 evaluator: Rc::clone(&evaluator),
1878 accumulator: Some(Box::new(SumAccumulator::default())),
1879 })
1880 }
1881 }
1882 AggregateFunction::Min => {
1883 let evaluator =
1884 self.expression_evaluator(expr, encoded_variables, stat_children)?;
1885 if *distinct {
1886 Box::new(move || AccumulatorWrapper::DistinctExpression {
1887 evaluator: Rc::clone(&evaluator),
1888 seen: FxHashSet::default(),
1889 accumulator: Some(Box::new(MinAccumulator::default())),
1890 })
1891 } else {
1892 Box::new(move || AccumulatorWrapper::Expression {
1893 evaluator: Rc::clone(&evaluator),
1894 accumulator: Some(Box::new(MinAccumulator::default())),
1895 })
1896 }
1897 }
1898 AggregateFunction::Max => {
1899 let evaluator =
1900 self.expression_evaluator(expr, encoded_variables, stat_children)?;
1901 if *distinct {
1902 Box::new(move || AccumulatorWrapper::DistinctExpression {
1903 evaluator: Rc::clone(&evaluator),
1904 seen: FxHashSet::default(),
1905 accumulator: Some(Box::new(MaxAccumulator::default())),
1906 })
1907 } else {
1908 Box::new(move || AccumulatorWrapper::Expression {
1909 evaluator: Rc::clone(&evaluator),
1910 accumulator: Some(Box::new(MaxAccumulator::default())),
1911 })
1912 }
1913 }
1914 AggregateFunction::Avg => {
1915 let evaluator =
1916 self.expression_evaluator(expr, encoded_variables, stat_children)?;
1917 if *distinct {
1918 Box::new(move || AccumulatorWrapper::DistinctExpression {
1919 evaluator: Rc::clone(&evaluator),
1920 seen: FxHashSet::default(),
1921 accumulator: Some(Box::new(AvgAccumulator::default())),
1922 })
1923 } else {
1924 Box::new(move || AccumulatorWrapper::Expression {
1925 evaluator: Rc::clone(&evaluator),
1926 accumulator: Some(Box::new(AvgAccumulator::default())),
1927 })
1928 }
1929 }
1930 AggregateFunction::Sample => {
1931 let evaluator =
1932 self.expression_evaluator(expr, encoded_variables, stat_children)?;
1933 Box::new(move || AccumulatorWrapper::Sample {
1934 evaluator: Rc::clone(&evaluator),
1935 value: None,
1936 })
1937 }
1938 AggregateFunction::GroupConcat { separator } => {
1939 let separator = Rc::from(separator.as_deref().unwrap_or(" "));
1940 let evaluator =
1941 self.expression_evaluator(expr, encoded_variables, stat_children)?;
1942 if *distinct {
1943 Box::new(move || AccumulatorWrapper::DistinctExpression {
1944 evaluator: Rc::clone(&evaluator),
1945 seen: FxHashSet::default(),
1946 accumulator: Some(Box::new(GroupConcatAccumulator::new(Rc::clone(
1947 &separator,
1948 )))),
1949 })
1950 } else {
1951 Box::new(move || AccumulatorWrapper::Expression {
1952 evaluator: Rc::clone(&evaluator),
1953 accumulator: Some(Box::new(GroupConcatAccumulator::new(Rc::clone(
1954 &separator,
1955 )))),
1956 })
1957 }
1958 }
1959 AggregateFunction::Custom(function_name) => {
1960 let Some(function) = self.custom_aggregate_functions.get(function_name) else {
1961 return Err(QueryEvaluationError::UnsupportedCustomFunction(
1962 function_name.clone(),
1963 ));
1964 };
1965 let evaluator =
1966 self.expression_evaluator(expr, encoded_variables, stat_children)?;
1967 let function = Arc::clone(function);
1968 if *distinct {
1969 Box::new(move || AccumulatorWrapper::DistinctExpression {
1970 evaluator: Rc::clone(&evaluator),
1971 seen: FxHashSet::default(),
1972 accumulator: Some(Box::new(CustomAccumulator(function()))),
1973 })
1974 } else {
1975 Box::new(move || AccumulatorWrapper::Expression {
1976 evaluator: Rc::clone(&evaluator),
1977 accumulator: Some(Box::new(CustomAccumulator(function()))),
1978 })
1979 }
1980 }
1981 },
1982 })
1983 }
1984
1985 #[expect(clippy::type_complexity)]
1989 fn internal_expression_evaluator(
1990 &self,
1991 expression: &Expression,
1992 encoded_variables: &mut Vec<Variable>,
1993 stat_children: &mut Vec<Rc<EvalNodeWithStats>>,
1994 ) -> Result<
1995 Option<Rc<dyn Fn(&InternalTuple<D::InternalTerm>) -> Option<D::InternalTerm> + 'a>>,
1996 QueryEvaluationError,
1997 > {
1998 Ok(try_build_internal_expression_evaluator(
1999 expression,
2000 &mut ExpressionContext {
2001 evaluator: self,
2002 encoded_variables,
2003 stat_children,
2004 },
2005 )?)
2006 }
2007
2008 pub(crate) fn effective_boolean_value_expression_evaluator(
2010 &self,
2011 expression: &Expression,
2012 encoded_variables: &mut Vec<Variable>,
2013 stat_children: &mut Vec<Rc<EvalNodeWithStats>>,
2014 ) -> Result<ExpressionEvaluator<'a, InternalTuple<D::InternalTerm>, bool>, QueryEvaluationError>
2015 {
2016 if let Some(eval) =
2018 self.internal_expression_evaluator(expression, encoded_variables, stat_children)?
2019 {
2020 let dataset = self.dataset.clone();
2021 return Ok(Rc::new(move |tuple| {
2022 dataset
2023 .internal_term_effective_boolean_value(eval(tuple)?)
2024 .ok()?
2025 }));
2026 }
2027 let eval = self.expression_evaluator(expression, encoded_variables, stat_children)?;
2028 Ok(Rc::new(move |tuple| eval(tuple)?.effective_boolean_value()))
2029 }
2030
2031 pub(crate) fn expression_evaluator(
2033 &self,
2034 expression: &Expression,
2035 encoded_variables: &mut Vec<Variable>,
2036 stat_children: &mut Vec<Rc<EvalNodeWithStats>>,
2037 ) -> Result<
2038 ExpressionEvaluator<'a, InternalTuple<D::InternalTerm>, ExpressionTerm>,
2039 QueryEvaluationError,
2040 > {
2041 Ok(build_expression_evaluator(
2042 expression,
2043 &mut ExpressionContext {
2044 evaluator: self,
2045 encoded_variables,
2046 stat_children,
2047 },
2048 )?)
2049 }
2050
2051 fn encode_term(&self, term: impl Into<Term>) -> Result<D::InternalTerm, QueryEvaluationError> {
2052 self.dataset.internalize_term(term.into())
2053 }
2054
2055 #[cfg(feature = "sparql-12")]
2056 fn encode_triple(
2057 &self,
2058 triple: &GroundTriple,
2059 ) -> Result<D::InternalTerm, QueryEvaluationError> {
2060 self.dataset.internalize_expression_term(
2061 ExpressionTriple::from(Triple::from(triple.clone())).into(),
2062 )
2063 }
2064
2065 fn encode_property_path(
2066 &self,
2067 path: &PropertyPathExpression,
2068 ) -> Result<Rc<PropertyPath<D::InternalTerm>>, QueryEvaluationError> {
2069 Ok(Rc::new(match path {
2070 PropertyPathExpression::NamedNode(node) => {
2071 PropertyPath::Path(self.encode_term(node.clone())?)
2072 }
2073 PropertyPathExpression::Reverse(p) => {
2074 PropertyPath::Reverse(self.encode_property_path(p)?)
2075 }
2076 PropertyPathExpression::Sequence(a, b) => {
2077 PropertyPath::Sequence(self.encode_property_path(a)?, self.encode_property_path(b)?)
2078 }
2079 PropertyPathExpression::Alternative(a, b) => PropertyPath::Alternative(
2080 self.encode_property_path(a)?,
2081 self.encode_property_path(b)?,
2082 ),
2083 PropertyPathExpression::ZeroOrMore(p) => {
2084 PropertyPath::ZeroOrMore(self.encode_property_path(p)?)
2085 }
2086 PropertyPathExpression::OneOrMore(p) => {
2087 PropertyPath::OneOrMore(self.encode_property_path(p)?)
2088 }
2089 PropertyPathExpression::ZeroOrOne(p) => {
2090 PropertyPath::ZeroOrOne(self.encode_property_path(p)?)
2091 }
2092 PropertyPathExpression::NegatedPropertySet(ps) => PropertyPath::NegatedPropertySet(
2093 ps.iter()
2094 .map(|p| self.encode_term(p.clone()))
2095 .collect::<Result<Rc<[_]>, _>>()?,
2096 ),
2097 }))
2098 }
2099}
2100
2101impl<'a, D: QueryableDataset<'a>> Clone for SimpleEvaluator<'a, D> {
2102 fn clone(&self) -> Self {
2103 Self {
2104 dataset: self.dataset.clone(),
2105 base_iri: self.base_iri.clone(),
2106 now: self.now,
2107 service_handler: Rc::clone(&self.service_handler),
2108 custom_functions: Rc::clone(&self.custom_functions),
2109 custom_aggregate_functions: Rc::clone(&self.custom_aggregate_functions),
2110 run_stats: self.run_stats,
2111 }
2112 }
2113}
2114
2115struct ExpressionContext<'a, E> {
2116 evaluator: &'a E,
2117 encoded_variables: &'a mut Vec<Variable>,
2118 stat_children: &'a mut Vec<Rc<EvalNodeWithStats>>,
2119}
2120
2121impl<'a, D: QueryableDataset<'a>> ExpressionEvaluatorContext<'a>
2122 for ExpressionContext<'_, SimpleEvaluator<'a, D>>
2123{
2124 type Tuple = InternalTuple<D::InternalTerm>;
2125 type Term = D::InternalTerm;
2126 type Error = QueryEvaluationError;
2127
2128 fn build_variable_lookup(
2129 &mut self,
2130 variable: &Variable,
2131 ) -> impl Fn(&InternalTuple<D::InternalTerm>) -> Option<D::InternalTerm> + 'a {
2132 let variable = encode_variable(self.encoded_variables, variable);
2133 move |tuple| tuple.get(variable).cloned()
2134 }
2135
2136 fn build_is_variable_bound(
2137 &mut self,
2138 variable: &Variable,
2139 ) -> impl Fn(&InternalTuple<D::InternalTerm>) -> bool + 'a {
2140 let variable = encode_variable(self.encoded_variables, variable);
2141 move |tuple| tuple.contains(variable)
2142 }
2143
2144 fn build_exists(
2145 &mut self,
2146 plan: &GraphPattern,
2147 ) -> Result<impl Fn(&InternalTuple<D::InternalTerm>) -> bool + 'a, QueryEvaluationError> {
2148 let (eval, stats) = self
2149 .evaluator
2150 .graph_pattern_evaluator(plan, self.encoded_variables);
2151 self.stat_children.push(stats);
2152 let eval = eval?;
2153 Ok(move |tuple: &InternalTuple<D::InternalTerm>| eval(tuple.clone()).next().is_some())
2154 }
2155
2156 fn internalize_named_node(
2157 &mut self,
2158 term: &NamedNode,
2159 ) -> Result<Self::Term, QueryEvaluationError> {
2160 self.evaluator.encode_term(term.clone())
2161 }
2162
2163 fn internalize_literal(&mut self, term: &Literal) -> Result<Self::Term, QueryEvaluationError> {
2164 self.evaluator.encode_term(term.clone())
2165 }
2166
2167 fn build_internalize_expression_term(
2168 &mut self,
2169 ) -> impl Fn(ExpressionTerm) -> Option<Self::Term> + 'a {
2170 let dataset = self.evaluator.dataset.clone();
2171 move |t| dataset.internalize_expression_term(t).ok()
2172 }
2173
2174 fn build_externalize_expression_term(
2175 &mut self,
2176 ) -> impl Fn(Self::Term) -> Option<ExpressionTerm> + 'a {
2177 let dataset = self.evaluator.dataset.clone();
2178 move |t| dataset.externalize_expression_term(t).ok()
2179 }
2180
2181 fn now(&mut self) -> DateTime {
2182 self.evaluator.now
2183 }
2184
2185 fn base_iri(&mut self) -> Option<Arc<Iri<String>>> {
2186 self.evaluator.base_iri.as_ref().map(Arc::clone)
2187 }
2188
2189 fn custom_functions(&mut self) -> &CustomFunctionRegistry {
2190 &self.evaluator.custom_functions
2191 }
2192}
2193
2194#[cfg(feature = "sparql-12")]
2195type LanguageWithMaybeBaseDirection = (String, Option<BaseDirection>);
2196#[cfg(not(feature = "sparql-12"))]
2197type LanguageWithMaybeBaseDirection = String;
2198
2199#[cfg(feature = "sparql-12")]
2200fn to_string_and_language(
2201 term: ExpressionTerm,
2202) -> Option<(String, Option<LanguageWithMaybeBaseDirection>)> {
2203 match term {
2204 ExpressionTerm::StringLiteral(value) => Some((value, None)),
2205 ExpressionTerm::LangStringLiteral { value, language } => {
2206 Some((value, Some((language, None))))
2207 }
2208 ExpressionTerm::DirLangStringLiteral {
2209 value,
2210 language,
2211 direction,
2212 } => Some((value, Some((language, Some(direction))))),
2213 _ => None,
2214 }
2215}
2216
2217#[cfg(not(feature = "sparql-12"))]
2218fn to_string_and_language(
2219 term: ExpressionTerm,
2220) -> Option<(String, Option<LanguageWithMaybeBaseDirection>)> {
2221 match term {
2222 ExpressionTerm::StringLiteral(value) => Some((value, None)),
2223 ExpressionTerm::LangStringLiteral { value, language } => Some((value, Some(language))),
2224 _ => None,
2225 }
2226}
2227
2228#[cfg(feature = "sparql-12")]
2229fn build_plain_literal(
2230 value: String,
2231 language: Option<LanguageWithMaybeBaseDirection>,
2232) -> ExpressionTerm {
2233 if let Some((language, direction)) = language {
2234 if let Some(direction) = direction {
2235 ExpressionTerm::DirLangStringLiteral {
2236 value,
2237 language,
2238 direction,
2239 }
2240 } else {
2241 ExpressionTerm::LangStringLiteral { value, language }
2242 }
2243 } else {
2244 ExpressionTerm::StringLiteral(value)
2245 }
2246}
2247
2248#[cfg(not(feature = "sparql-12"))]
2249fn build_plain_literal(
2250 value: String,
2251 language: Option<LanguageWithMaybeBaseDirection>,
2252) -> ExpressionTerm {
2253 if let Some(language) = language {
2254 ExpressionTerm::LangStringLiteral { value, language }
2255 } else {
2256 ExpressionTerm::StringLiteral(value)
2257 }
2258}
2259
2260fn decode_bindings<'a, D: QueryableDataset<'a>>(
2261 dataset: EvalDataset<'a, D>,
2262 iter: InternalTuplesIterator<'a, D::InternalTerm>,
2263 variables: Arc<[Variable]>,
2264) -> QuerySolutionIter<'a> {
2265 let tuple_size = variables.len();
2266 QuerySolutionIter::from_tuples(
2267 variables,
2268 Box::new(iter.map(move |values| {
2269 let mut result = vec![None; tuple_size];
2270 for (i, value) in values?.iter().enumerate() {
2271 if let Some(term) = value {
2272 result[i] = Some(dataset.externalize_term(term)?)
2273 }
2274 }
2275 Ok(result)
2276 })),
2277 )
2278}
2279
2280fn encode_bindings<'a, D: QueryableDataset<'a>>(
2282 dataset: EvalDataset<'a, D>,
2283 variables: Rc<[Variable]>,
2284 iter: QuerySolutionIter<'a>,
2285) -> InternalTuplesIterator<'a, D::InternalTerm> {
2286 Box::new(iter.map(move |solution| {
2287 let mut encoded_terms = InternalTuple::with_capacity(variables.len());
2288 for (variable, term) in &solution? {
2289 put_variable_value(
2290 variable,
2291 &variables,
2292 dataset.internalize_term(term.clone())?,
2293 &mut encoded_terms,
2294 );
2295 }
2296 Ok(encoded_terms)
2297 }))
2298}
2299
2300fn encode_initial_bindings<'a, D: QueryableDataset<'a>>(
2301 dataset: &EvalDataset<'a, D>,
2302 variables: &[Variable],
2303 values: impl IntoIterator<Item = (Variable, Term)>,
2304) -> Result<InternalTuple<D::InternalTerm>, QueryEvaluationError> {
2305 let mut encoded_terms = InternalTuple::with_capacity(variables.len());
2306 for (variable, term) in values {
2307 if !put_variable_value(
2308 &variable,
2309 variables,
2310 dataset.internalize_term(term)?,
2311 &mut encoded_terms,
2312 ) {
2313 return Err(QueryEvaluationError::NotExistingSubstitutedVariable(
2314 variable,
2315 ));
2316 }
2317 }
2318 Ok(encoded_terms)
2319}
2320
2321fn put_variable_value<T: Clone>(
2322 selector: &Variable,
2323 variables: &[Variable],
2324 value: T,
2325 tuple: &mut InternalTuple<T>,
2326) -> bool {
2327 for (i, v) in variables.iter().enumerate() {
2328 if selector == v {
2329 tuple.set(i, value);
2330 return true;
2331 }
2332 }
2333 false
2334}
2335
2336enum AccumulatorWrapper<'a, T> {
2337 CountTuple {
2338 count: u64,
2339 },
2340 CountDistinctTuple {
2341 seen: FxHashSet<InternalTuple<T>>,
2342 count: u64,
2343 },
2344 CountInternal {
2345 evaluator: Rc<dyn Fn(&InternalTuple<T>) -> Option<T> + 'a>,
2346 count: u64,
2347 },
2348 CountDistinctInternal {
2349 seen: FxHashSet<T>,
2350 evaluator: Rc<dyn Fn(&InternalTuple<T>) -> Option<T> + 'a>,
2351 count: u64,
2352 },
2353 Sample {
2354 evaluator: Rc<dyn Fn(&InternalTuple<T>) -> Option<ExpressionTerm> + 'a>,
2356 value: Option<ExpressionTerm>,
2357 },
2358 Expression {
2359 evaluator: Rc<dyn Fn(&InternalTuple<T>) -> Option<ExpressionTerm> + 'a>,
2360 accumulator: Option<Box<dyn Accumulator>>,
2361 },
2362 DistinctExpression {
2363 seen: FxHashSet<ExpressionTerm>,
2364 evaluator: Rc<dyn Fn(&InternalTuple<T>) -> Option<ExpressionTerm> + 'a>,
2365 accumulator: Option<Box<dyn Accumulator>>,
2366 },
2367}
2368
2369impl<T: Clone + Eq + Hash> AccumulatorWrapper<'_, T> {
2370 fn accumulate(&mut self, tuple: &InternalTuple<T>) {
2371 match self {
2372 Self::CountTuple { count } => {
2373 *count += 1;
2374 }
2375 Self::CountDistinctTuple { seen, count } => {
2376 if seen.insert(tuple.clone()) {
2377 *count += 1;
2378 }
2379 }
2380 Self::CountInternal { evaluator, count } => {
2381 if evaluator(tuple).is_some() {
2382 *count += 1;
2383 }
2384 }
2385 Self::CountDistinctInternal {
2386 seen,
2387 evaluator,
2388 count,
2389 } => {
2390 let Some(value) = evaluator(tuple) else {
2391 return;
2392 };
2393 if seen.insert(value) {
2394 *count += 1;
2395 }
2396 }
2397 Self::Sample { evaluator, value } => {
2398 if value.is_some() {
2399 return; }
2401 *value = evaluator(tuple);
2402 }
2403 Self::Expression {
2404 evaluator,
2405 accumulator,
2406 } => {
2407 if accumulator.is_none() {
2408 return; }
2410 let Some(value) = evaluator(tuple) else {
2411 *accumulator = None;
2412 return;
2413 };
2414 let Some(accumulator) = accumulator else {
2415 return;
2416 };
2417 accumulator.accumulate(value);
2418 }
2419 Self::DistinctExpression {
2420 seen,
2421 evaluator,
2422 accumulator,
2423 } => {
2424 if accumulator.is_none() {
2425 return; }
2427 let Some(value) = evaluator(tuple) else {
2428 *accumulator = None;
2429 return;
2430 };
2431 let Some(accumulator) = accumulator else {
2432 return;
2433 };
2434 if seen.insert(value.clone()) {
2435 accumulator.accumulate(value);
2436 }
2437 }
2438 }
2439 }
2440
2441 fn finish(self) -> Option<ExpressionTerm> {
2442 match self {
2443 Self::CountTuple { count, .. }
2444 | Self::CountDistinctTuple { count, .. }
2445 | Self::CountInternal { count, .. }
2446 | Self::CountDistinctInternal { count, .. } => Some(ExpressionTerm::IntegerLiteral(
2447 i64::try_from(count).ok()?.into(),
2448 )),
2449 Self::Sample { value, .. } => value,
2450 Self::Expression { accumulator, .. } | Self::DistinctExpression { accumulator, .. } => {
2451 accumulator?.finish()
2452 }
2453 }
2454 }
2455}
2456
2457trait Accumulator {
2458 fn accumulate(&mut self, element: ExpressionTerm);
2459
2460 fn finish(&mut self) -> Option<ExpressionTerm>;
2461}
2462
2463#[derive(Default, Debug)]
2464struct CountAccumulator {
2465 count: i64,
2466}
2467
2468impl Accumulator for CountAccumulator {
2469 fn accumulate(&mut self, _element: ExpressionTerm) {
2470 self.count += 1;
2471 }
2472
2473 fn finish(&mut self) -> Option<ExpressionTerm> {
2474 Some(ExpressionTerm::IntegerLiteral(self.count.into()))
2475 }
2476}
2477
2478struct SumAccumulator {
2479 sum: Option<ExpressionTerm>,
2480}
2481
2482impl Default for SumAccumulator {
2483 fn default() -> Self {
2484 Self {
2485 sum: Some(ExpressionTerm::IntegerLiteral(Integer::default())),
2486 }
2487 }
2488}
2489
2490impl Accumulator for SumAccumulator {
2491 fn accumulate(&mut self, element: ExpressionTerm) {
2492 let Some(sum) = &self.sum else {
2493 return;
2494 };
2495 self.sum = if let Some(operands) = NumericBinaryOperands::new(sum.clone(), element) {
2496 match operands {
2498 NumericBinaryOperands::Float(v1, v2) => Some(ExpressionTerm::FloatLiteral(v1 + v2)),
2499 NumericBinaryOperands::Double(v1, v2) => {
2500 Some(ExpressionTerm::DoubleLiteral(v1 + v2))
2501 }
2502 NumericBinaryOperands::Integer(v1, v2) => {
2503 v1.checked_add(v2).map(ExpressionTerm::IntegerLiteral)
2504 }
2505 NumericBinaryOperands::Decimal(v1, v2) => {
2506 v1.checked_add(v2).map(ExpressionTerm::DecimalLiteral)
2507 }
2508 #[cfg(feature = "sep-0002")]
2509 _ => None,
2510 }
2511 } else {
2512 None
2513 };
2514 }
2515
2516 fn finish(&mut self) -> Option<ExpressionTerm> {
2517 self.sum.take()
2518 }
2519}
2520
2521#[derive(Default)]
2522struct AvgAccumulator {
2523 sum: SumAccumulator,
2524 count: i64,
2525}
2526
2527impl Accumulator for AvgAccumulator {
2528 fn accumulate(&mut self, element: ExpressionTerm) {
2529 self.sum.accumulate(element);
2530 self.count += 1;
2531 }
2532
2533 fn finish(&mut self) -> Option<ExpressionTerm> {
2534 let sum = self.sum.finish()?;
2535 if self.count == 0 {
2536 return Some(ExpressionTerm::IntegerLiteral(0.into()));
2537 }
2538 let count = Integer::from(self.count);
2540 match sum {
2541 ExpressionTerm::FloatLiteral(sum) => {
2542 Some(ExpressionTerm::FloatLiteral(sum / Float::from(count)))
2543 }
2544 ExpressionTerm::DoubleLiteral(sum) => {
2545 Some(ExpressionTerm::DoubleLiteral(sum / Double::from(count)))
2546 }
2547 ExpressionTerm::IntegerLiteral(sum) => Some(ExpressionTerm::DecimalLiteral(
2548 Decimal::from(sum).checked_div(count)?,
2549 )),
2550 ExpressionTerm::DecimalLiteral(sum) => {
2551 Some(ExpressionTerm::DecimalLiteral(sum.checked_div(count)?))
2552 }
2553 _ => None,
2554 }
2555 }
2556}
2557
2558#[derive(Default)]
2559#[expect(clippy::option_option)]
2560struct MinAccumulator {
2561 min: Option<Option<ExpressionTerm>>,
2562}
2563
2564impl Accumulator for MinAccumulator {
2565 fn accumulate(&mut self, element: ExpressionTerm) {
2566 if let Some(min) = &self.min {
2567 if cmp_terms(Some(&element), min.as_ref()) == Ordering::Less {
2568 self.min = Some(Some(element));
2569 }
2570 } else {
2571 self.min = Some(Some(element))
2572 }
2573 }
2574
2575 fn finish(&mut self) -> Option<ExpressionTerm> {
2576 self.min.clone()?
2577 }
2578}
2579
2580#[derive(Default)]
2581#[expect(clippy::option_option)]
2582struct MaxAccumulator {
2583 max: Option<Option<ExpressionTerm>>,
2584}
2585
2586impl Accumulator for MaxAccumulator {
2587 fn accumulate(&mut self, element: ExpressionTerm) {
2588 if let Some(max) = &self.max {
2589 if cmp_terms(Some(&element), max.as_ref()) == Ordering::Greater {
2590 self.max = Some(Some(element))
2591 }
2592 } else {
2593 self.max = Some(Some(element))
2594 }
2595 }
2596
2597 fn finish(&mut self) -> Option<ExpressionTerm> {
2598 self.max.clone()?
2599 }
2600}
2601
2602#[expect(clippy::option_option)]
2603struct GroupConcatAccumulator {
2604 concat: Option<String>,
2605 language: Option<Option<LanguageWithMaybeBaseDirection>>,
2606 separator: Rc<str>,
2607}
2608
2609impl GroupConcatAccumulator {
2610 fn new(separator: Rc<str>) -> Self {
2611 Self {
2612 concat: Some(String::new()),
2613 language: None,
2614 separator,
2615 }
2616 }
2617}
2618
2619impl Accumulator for GroupConcatAccumulator {
2620 fn accumulate(&mut self, element: ExpressionTerm) {
2621 let Some(concat) = self.concat.as_mut() else {
2622 return;
2623 };
2624 let Some((value, e_language)) = to_string_and_language(element) else {
2625 self.concat = None;
2626 return;
2627 };
2628 if let Some(lang) = &self.language {
2629 if *lang != e_language {
2630 self.language = Some(None)
2631 }
2632 concat.push_str(&self.separator);
2633 } else {
2634 self.language = Some(e_language)
2635 }
2636 concat.push_str(&value);
2637 }
2638
2639 fn finish(&mut self) -> Option<ExpressionTerm> {
2640 self.concat
2641 .take()
2642 .map(|result| build_plain_literal(result, self.language.take().flatten()))
2643 }
2644}
2645
2646struct CustomAccumulator(Box<dyn AggregateFunctionAccumulator + Send + Sync>);
2647
2648impl Accumulator for CustomAccumulator {
2649 fn accumulate(&mut self, element: ExpressionTerm) {
2650 self.0.accumulate(element.into())
2651 }
2652
2653 fn finish(&mut self) -> Option<ExpressionTerm> {
2654 Some(self.0.finish()?.into())
2655 }
2656}
2657
2658fn encode_variable(variables: &mut Vec<Variable>, variable: &Variable) -> usize {
2659 if let Some(key) = slice_key(variables, variable) {
2660 key
2661 } else {
2662 variables.push(variable.clone());
2663 variables.len() - 1
2664 }
2665}
2666
2667fn bnode_key(blank_nodes: &mut Vec<BlankNode>, blank_node: &BlankNode) -> usize {
2668 if let Some(key) = slice_key(blank_nodes, blank_node) {
2669 key
2670 } else {
2671 blank_nodes.push(blank_node.clone());
2672 blank_nodes.len() - 1
2673 }
2674}
2675
2676fn slice_key<T: Eq>(slice: &[T], element: &T) -> Option<usize> {
2677 for (i, item) in slice.iter().enumerate() {
2678 if item == element {
2679 return Some(i);
2680 }
2681 }
2682 None
2683}
2684
2685fn cmp_terms(a: Option<&ExpressionTerm>, b: Option<&ExpressionTerm>) -> Ordering {
2687 match (a, b) {
2688 (Some(a), Some(b)) => {
2689 match a {
2690 ExpressionTerm::BlankNode(a) => match b {
2691 ExpressionTerm::BlankNode(b) => a.as_str().cmp(b.as_str()),
2692 _ => Ordering::Less,
2693 },
2694 ExpressionTerm::NamedNode(a) => match b {
2695 ExpressionTerm::BlankNode(_) => Ordering::Greater,
2696 ExpressionTerm::NamedNode(b) => a.as_str().cmp(b.as_str()),
2697 _ => Ordering::Less,
2698 },
2699 #[cfg(feature = "sparql-12")]
2700 ExpressionTerm::Triple(a) => match b {
2701 ExpressionTerm::Triple(b) => cmp_triples(a, b),
2702 _ => Ordering::Greater,
2703 },
2704 _ => match b {
2705 ExpressionTerm::NamedNode(_) | ExpressionTerm::BlankNode(_) => {
2706 Ordering::Greater
2707 }
2708 #[cfg(feature = "sparql-12")]
2709 ExpressionTerm::Triple(_) => Ordering::Less,
2710 _ => {
2711 if let Some(ord) = partial_cmp_literals(a, b) {
2712 ord
2713 } else if let (Term::Literal(a), Term::Literal(b)) =
2714 (a.clone().into(), b.clone().into())
2715 {
2716 (a.value(), a.datatype(), a.language()).cmp(&(
2717 b.value(),
2718 b.datatype(),
2719 b.language(),
2720 ))
2721 } else {
2722 Ordering::Equal }
2724 }
2725 },
2726 }
2727 }
2728 (Some(_), None) => Ordering::Greater,
2729 (None, Some(_)) => Ordering::Less,
2730 (None, None) => Ordering::Equal,
2731 }
2732}
2733
2734#[cfg(feature = "sparql-12")]
2735fn cmp_triples(a: &ExpressionTriple, b: &ExpressionTriple) -> Ordering {
2736 match match &a.subject {
2737 NamedOrBlankNode::BlankNode(a) => match &b.subject {
2738 NamedOrBlankNode::BlankNode(b) => a.as_str().cmp(b.as_str()),
2739 NamedOrBlankNode::NamedNode(_) => Ordering::Less,
2740 },
2741 NamedOrBlankNode::NamedNode(a) => match &b.subject {
2742 NamedOrBlankNode::BlankNode(_) => Ordering::Greater,
2743 NamedOrBlankNode::NamedNode(b) => a.as_str().cmp(b.as_str()),
2744 },
2745 } {
2746 Ordering::Equal => match a.predicate.as_str().cmp(b.predicate.as_str()) {
2747 Ordering::Equal => cmp_terms(Some(&a.object), Some(&b.object)),
2748 o => o,
2749 },
2750 o => o,
2751 }
2752}
2753
2754enum TupleSelector<T> {
2755 Constant(T),
2756 Variable(usize),
2757 #[cfg(feature = "sparql-12")]
2758 TriplePattern(Rc<TripleTupleSelector<T>>),
2759}
2760
2761impl<T> TupleSelector<T> {
2762 fn from_ground_term_pattern<'a>(
2763 term_pattern: &GroundTermPattern,
2764 variables: &mut Vec<Variable>,
2765 dataset: &EvalDataset<'a, impl QueryableDataset<'a, InternalTerm = T>>,
2766 ) -> Result<Self, QueryEvaluationError> {
2767 Ok(match term_pattern {
2768 GroundTermPattern::Variable(variable) => {
2769 Self::Variable(encode_variable(variables, variable))
2770 }
2771 GroundTermPattern::NamedNode(term) => {
2772 Self::Constant(dataset.internalize_term(term.as_ref().into())?)
2773 }
2774 GroundTermPattern::Literal(term) => {
2775 Self::Constant(dataset.internalize_term(term.as_ref().into())?)
2776 }
2777 #[cfg(feature = "sparql-12")]
2778 GroundTermPattern::Triple(triple) => {
2779 match (
2780 Self::from_ground_term_pattern(&triple.subject, variables, dataset)?,
2781 Self::from_named_node_pattern(&triple.predicate, variables, dataset)?,
2782 Self::from_ground_term_pattern(&triple.object, variables, dataset)?,
2783 ) {
2784 (
2785 Self::Constant(subject),
2786 Self::Constant(predicate),
2787 Self::Constant(object),
2788 ) => Self::Constant(
2789 dataset.internalize_expression_term(
2790 ExpressionTriple::new(
2791 dataset.externalize_expression_term(subject)?,
2792 dataset.externalize_expression_term(predicate)?,
2793 dataset.externalize_expression_term(object)?,
2794 )
2795 .ok_or_else(|| QueryEvaluationError::InvalidStorageTripleTerm)?
2796 .into(),
2797 )?,
2798 ),
2799 (subject, predicate, object) => {
2800 Self::TriplePattern(Rc::new(TripleTupleSelector {
2801 subject,
2802 predicate,
2803 object,
2804 }))
2805 }
2806 }
2807 }
2808 })
2809 }
2810
2811 fn from_named_node_pattern<'a>(
2812 named_node_pattern: &NamedNodePattern,
2813 variables: &mut Vec<Variable>,
2814 dataset: &EvalDataset<'a, impl QueryableDataset<'a, InternalTerm = T>>,
2815 ) -> Result<Self, QueryEvaluationError> {
2816 Ok(match named_node_pattern {
2817 NamedNodePattern::Variable(variable) => {
2818 Self::Variable(encode_variable(variables, variable))
2819 }
2820 NamedNodePattern::NamedNode(term) => {
2821 Self::Constant(dataset.internalize_term(term.as_ref().into())?)
2822 }
2823 })
2824 }
2825}
2826
2827impl<T: Clone> TupleSelector<T> {
2828 #[cfg_attr(
2829 not(feature = "sparql-12"),
2830 expect(
2831 unused_lifetimes,
2832 clippy::unnecessary_wraps,
2833 clippy::extra_unused_lifetimes
2834 )
2835 )]
2836 fn get_pattern_value<'a>(
2837 &self,
2838 tuple: &InternalTuple<T>,
2839 #[cfg(feature = "sparql-12")] dataset: &EvalDataset<
2840 'a,
2841 impl QueryableDataset<'a, InternalTerm = T>,
2842 >,
2843 ) -> Result<Option<T>, QueryEvaluationError> {
2844 Ok(match self {
2845 Self::Constant(c) => Some(c.clone()),
2846 Self::Variable(v) => tuple.get(*v).cloned(),
2847 #[cfg(feature = "sparql-12")]
2848 Self::TriplePattern(triple) => {
2849 let Some(subject) = triple.subject.get_pattern_value(tuple, dataset)? else {
2850 return Ok(None);
2851 };
2852 let Some(predicate) = triple.predicate.get_pattern_value(tuple, dataset)? else {
2853 return Ok(None);
2854 };
2855 let Some(object) = triple.object.get_pattern_value(tuple, dataset)? else {
2856 return Ok(None);
2857 };
2858 Some(
2859 dataset.internalize_expression_term(
2860 ExpressionTriple::new(
2861 dataset.externalize_expression_term(subject)?,
2862 dataset.externalize_expression_term(predicate)?,
2863 dataset.externalize_expression_term(object)?,
2864 )
2865 .ok_or(QueryEvaluationError::InvalidStorageTripleTerm)?
2866 .into(),
2867 )?,
2868 )
2869 }
2870 })
2871 }
2872}
2873
2874impl<T: Clone> Clone for TupleSelector<T> {
2875 fn clone(&self) -> Self {
2876 match self {
2877 Self::Constant(c) => Self::Constant(c.clone()),
2878 Self::Variable(v) => Self::Variable(*v),
2879 #[cfg(feature = "sparql-12")]
2880 Self::TriplePattern(t) => Self::TriplePattern(Rc::clone(t)),
2881 }
2882 }
2883}
2884
2885#[cfg(feature = "sparql-12")]
2886struct TripleTupleSelector<T> {
2887 subject: TupleSelector<T>,
2888 predicate: TupleSelector<T>,
2889 object: TupleSelector<T>,
2890}
2891
2892#[cfg_attr(not(feature = "sparql-12"), expect(clippy::unnecessary_wraps))]
2893fn put_pattern_value<'a, D: QueryableDataset<'a>>(
2894 selector: &TupleSelector<D::InternalTerm>,
2895 value: D::InternalTerm,
2896 tuple: &mut InternalTuple<D::InternalTerm>,
2897 #[cfg(feature = "sparql-12")] dataset: &EvalDataset<'a, D>,
2898) -> Result<bool, QueryEvaluationError> {
2899 Ok(match selector {
2900 TupleSelector::Constant(c) => *c == value,
2901 TupleSelector::Variable(v) => {
2902 if let Some(old) = tuple.get(*v) {
2903 value == *old
2904 } else {
2905 tuple.set(*v, value);
2906 true
2907 }
2908 }
2909 #[cfg(feature = "sparql-12")]
2910 TupleSelector::TriplePattern(triple) => {
2911 let ExpressionTerm::Triple(value) = dataset.externalize_expression_term(value)? else {
2912 return Ok(false);
2913 };
2914 put_pattern_value(
2915 &triple.subject,
2916 dataset.internalize_expression_term(value.subject.into())?,
2917 tuple,
2918 dataset,
2919 )? && put_pattern_value(
2920 &triple.predicate,
2921 dataset.internalize_expression_term(value.predicate.into())?,
2922 tuple,
2923 dataset,
2924 )? && put_pattern_value(
2925 &triple.object,
2926 dataset.internalize_expression_term(value.object)?,
2927 tuple,
2928 dataset,
2929 )?
2930 }
2931 })
2932}
2933
2934pub fn are_compatible_and_not_disjointed<T: Clone + Eq>(
2935 a: &InternalTuple<T>,
2936 b: &InternalTuple<T>,
2937) -> bool {
2938 let mut found_intersection = false;
2939 for (a_value, b_value) in a.iter().zip(b.iter()) {
2940 if let (Some(a_value), Some(b_value)) = (a_value, b_value) {
2941 if a_value != b_value {
2942 return false;
2943 }
2944 found_intersection = true;
2945 }
2946 }
2947 found_intersection
2948}
2949
2950pub enum PropertyPath<T> {
2951 Path(T),
2952 Reverse(Rc<Self>),
2953 Sequence(Rc<Self>, Rc<Self>),
2954 Alternative(Rc<Self>, Rc<Self>),
2955 ZeroOrMore(Rc<Self>),
2956 OneOrMore(Rc<Self>),
2957 ZeroOrOne(Rc<Self>),
2958 NegatedPropertySet(Rc<[T]>),
2959}
2960
2961struct PathEvaluator<'a, D: QueryableDataset<'a>> {
2962 dataset: EvalDataset<'a, D>,
2963}
2964
2965impl<'a, D: QueryableDataset<'a>> PathEvaluator<'a, D> {
2966 fn eval_closed_in_graph(
2967 &self,
2968 path: &PropertyPath<D::InternalTerm>,
2969 start: &D::InternalTerm,
2970 end: &D::InternalTerm,
2971 graph_name: Option<&D::InternalTerm>,
2972 ) -> Result<bool, QueryEvaluationError> {
2973 Ok(match path {
2974 PropertyPath::Path(p) => self
2975 .dataset
2976 .internal_quads_for_pattern(Some(start), Some(p), Some(end), Some(graph_name))
2977 .next()
2978 .transpose()?
2979 .is_some(),
2980 PropertyPath::Reverse(p) => self.eval_closed_in_graph(p, end, start, graph_name)?,
2981 PropertyPath::Sequence(a, b) => self
2982 .eval_from_in_graph(a, start, graph_name)
2983 .find_map(|middle| {
2984 middle
2985 .and_then(|middle| {
2986 Ok(self
2987 .eval_closed_in_graph(b, &middle, end, graph_name)?
2988 .then_some(()))
2989 })
2990 .transpose()
2991 })
2992 .transpose()?
2993 .is_some(),
2994 PropertyPath::Alternative(a, b) => {
2995 self.eval_closed_in_graph(a, start, end, graph_name)?
2996 || self.eval_closed_in_graph(b, start, end, graph_name)?
2997 }
2998 PropertyPath::ZeroOrMore(p) => {
2999 if start == end {
3000 self.is_subject_or_object_in_graph(start, graph_name)?
3001 } else {
3002 look_in_transitive_closure(
3003 self.eval_from_in_graph(p, start, graph_name),
3004 move |e| self.eval_from_in_graph(p, &e, graph_name),
3005 end,
3006 )?
3007 }
3008 }
3009 PropertyPath::OneOrMore(p) => look_in_transitive_closure(
3010 self.eval_from_in_graph(p, start, graph_name),
3011 move |e| self.eval_from_in_graph(p, &e, graph_name),
3012 end,
3013 )?,
3014 PropertyPath::ZeroOrOne(p) => {
3015 if start == end {
3016 self.is_subject_or_object_in_graph(start, graph_name)
3017 } else {
3018 self.eval_closed_in_graph(p, start, end, graph_name)
3019 }?
3020 }
3021 PropertyPath::NegatedPropertySet(ps) => self
3022 .dataset
3023 .internal_quads_for_pattern(Some(start), None, Some(end), Some(graph_name))
3024 .find_map(move |t| match t {
3025 Ok(t) => {
3026 if ps.contains(&t.predicate) {
3027 None
3028 } else {
3029 Some(Ok(()))
3030 }
3031 }
3032 Err(e) => Some(Err(e)),
3033 })
3034 .transpose()?
3035 .is_some(),
3036 })
3037 }
3038
3039 fn eval_closed_in_unknown_graph(
3040 &self,
3041 path: &PropertyPath<D::InternalTerm>,
3042 start: &D::InternalTerm,
3043 end: &D::InternalTerm,
3044 ) -> Box<dyn Iterator<Item = Result<Option<D::InternalTerm>, QueryEvaluationError>> + 'a> {
3045 match path {
3046 PropertyPath::Path(p) => Box::new(
3047 self.dataset
3048 .internal_quads_for_pattern(Some(start), Some(p), Some(end), None)
3049 .map(|t| Ok(t?.graph_name)),
3050 ),
3051 PropertyPath::Reverse(p) => self.eval_closed_in_unknown_graph(p, end, start),
3052 PropertyPath::Sequence(a, b) => {
3053 let eval = self.clone();
3054 let b = Rc::clone(b);
3055 let end = end.clone();
3056 Box::new(self.eval_from_in_unknown_graph(a, start).flat_map_ok(
3057 move |(middle, graph_name)| {
3058 eval.eval_closed_in_graph(&b, &middle, &end, graph_name.as_ref())
3059 .map(|is_found| is_found.then_some(graph_name))
3060 .transpose()
3061 },
3062 ))
3063 }
3064 PropertyPath::Alternative(a, b) => Box::new(hash_deduplicate(
3065 self.eval_closed_in_unknown_graph(a, start, end)
3066 .chain(self.eval_closed_in_unknown_graph(b, start, end)),
3067 )),
3068 PropertyPath::ZeroOrMore(p) => {
3069 let eval = self.clone();
3070 let start2 = start.clone();
3071 let end = end.clone();
3072 let p = Rc::clone(p);
3073 self.run_if_term_is_a_dataset_node(start, move |graph_name| {
3074 look_in_transitive_closure(
3075 Some(Ok(start2.clone())),
3076 |e| eval.eval_from_in_graph(&p, &e, graph_name.as_ref()),
3077 &end,
3078 )
3079 .map(|is_found| is_found.then_some(graph_name))
3080 .transpose()
3081 })
3082 }
3083 PropertyPath::OneOrMore(p) => {
3084 let eval = self.clone();
3085 let end = end.clone();
3086 let p = Rc::clone(p);
3087 Box::new(
3088 self.eval_from_in_unknown_graph(&p, start)
3089 .filter_map(move |r| {
3090 r.and_then(|(start, graph_name)| {
3091 look_in_transitive_closure(
3092 Some(Ok(start)),
3093 |e| eval.eval_from_in_graph(&p, &e, graph_name.as_ref()),
3094 &end,
3095 )
3096 .map(|is_found| is_found.then_some(graph_name))
3097 })
3098 .transpose()
3099 }),
3100 )
3101 }
3102 PropertyPath::ZeroOrOne(p) => {
3103 if start == end {
3104 self.run_if_term_is_a_dataset_node(start, |graph_name| Some(Ok(graph_name)))
3105 } else {
3106 let eval = self.clone();
3107 let start2 = start.clone();
3108 let end = end.clone();
3109 let p = Rc::clone(p);
3110 self.run_if_term_is_a_dataset_node(start, move |graph_name| {
3111 eval.eval_closed_in_graph(&p, &start2, &end, graph_name.as_ref())
3112 .map(|is_found| is_found.then_some(graph_name))
3113 .transpose()
3114 })
3115 }
3116 }
3117 PropertyPath::NegatedPropertySet(ps) => {
3118 let ps = Rc::clone(ps);
3119 Box::new(
3120 self.dataset
3121 .internal_quads_for_pattern(Some(start), None, Some(end), None)
3122 .filter_map(move |t| match t {
3123 Ok(t) => {
3124 if ps.contains(&t.predicate) {
3125 None
3126 } else {
3127 Some(Ok(t.graph_name))
3128 }
3129 }
3130 Err(e) => Some(Err(e)),
3131 }),
3132 )
3133 }
3134 }
3135 }
3136
3137 fn eval_from_in_graph(
3138 &self,
3139 path: &PropertyPath<D::InternalTerm>,
3140 start: &D::InternalTerm,
3141 graph_name: Option<&D::InternalTerm>,
3142 ) -> Box<dyn Iterator<Item = Result<D::InternalTerm, QueryEvaluationError>> + 'a> {
3143 match path {
3144 PropertyPath::Path(p) => Box::new(
3145 self.dataset
3146 .internal_quads_for_pattern(Some(start), Some(p), None, Some(graph_name))
3147 .map(|t| Ok(t?.object)),
3148 ),
3149 PropertyPath::Reverse(p) => self.eval_to_in_graph(p, start, graph_name),
3150 PropertyPath::Sequence(a, b) => {
3151 let eval = self.clone();
3152 let b = Rc::clone(b);
3153 let graph_name2 = graph_name.cloned();
3154 Box::new(
3155 self.eval_from_in_graph(a, start, graph_name)
3156 .flat_map_ok(move |middle| {
3157 eval.eval_from_in_graph(&b, &middle, graph_name2.as_ref())
3158 }),
3159 )
3160 }
3161 PropertyPath::Alternative(a, b) => Box::new(hash_deduplicate(
3162 self.eval_from_in_graph(a, start, graph_name)
3163 .chain(self.eval_from_in_graph(b, start, graph_name)),
3164 )),
3165 PropertyPath::ZeroOrMore(p) => {
3166 self.run_if_term_is_a_graph_node(start, graph_name, || {
3167 let eval = self.clone();
3168 let p = Rc::clone(p);
3169 let graph_name2 = graph_name.cloned();
3170 transitive_closure(Some(Ok(start.clone())), move |e| {
3171 eval.eval_from_in_graph(&p, &e, graph_name2.as_ref())
3172 })
3173 })
3174 }
3175 PropertyPath::OneOrMore(p) => {
3176 let eval = self.clone();
3177 let p = Rc::clone(p);
3178 let graph_name2 = graph_name.cloned();
3179 Box::new(transitive_closure(
3180 self.eval_from_in_graph(&p, start, graph_name),
3181 move |e| eval.eval_from_in_graph(&p, &e, graph_name2.as_ref()),
3182 ))
3183 }
3184 PropertyPath::ZeroOrOne(p) => {
3185 self.run_if_term_is_a_graph_node(start, graph_name, || {
3186 hash_deduplicate(
3187 once(Ok(start.clone()))
3188 .chain(self.eval_from_in_graph(p, start, graph_name)),
3189 )
3190 })
3191 }
3192 PropertyPath::NegatedPropertySet(ps) => {
3193 let ps = Rc::clone(ps);
3194 Box::new(
3195 self.dataset
3196 .internal_quads_for_pattern(Some(start), None, None, Some(graph_name))
3197 .filter_map(move |t| match t {
3198 Ok(t) => {
3199 if ps.contains(&t.predicate) {
3200 None
3201 } else {
3202 Some(Ok(t.object))
3203 }
3204 }
3205 Err(e) => Some(Err(e)),
3206 }),
3207 )
3208 }
3209 }
3210 }
3211
3212 fn eval_from_in_unknown_graph(
3213 &self,
3214 path: &PropertyPath<D::InternalTerm>,
3215 start: &D::InternalTerm,
3216 ) -> Box<
3217 dyn Iterator<
3218 Item = Result<(D::InternalTerm, Option<D::InternalTerm>), QueryEvaluationError>,
3219 > + 'a,
3220 > {
3221 match path {
3222 PropertyPath::Path(p) => Box::new(
3223 self.dataset
3224 .internal_quads_for_pattern(Some(start), Some(p), None, None)
3225 .map(|t| {
3226 let t = t?;
3227 Ok((t.object, t.graph_name))
3228 }),
3229 ),
3230 PropertyPath::Reverse(p) => self.eval_to_in_unknown_graph(p, start),
3231 PropertyPath::Sequence(a, b) => {
3232 let eval = self.clone();
3233 let b = Rc::clone(b);
3234 Box::new(self.eval_from_in_unknown_graph(a, start).flat_map_ok(
3235 move |(middle, graph_name)| {
3236 eval.eval_from_in_graph(&b, &middle, graph_name.as_ref())
3237 .map(move |end| Ok((end?, graph_name.clone())))
3238 },
3239 ))
3240 }
3241 PropertyPath::Alternative(a, b) => Box::new(hash_deduplicate(
3242 self.eval_from_in_unknown_graph(a, start)
3243 .chain(self.eval_from_in_unknown_graph(b, start)),
3244 )),
3245 PropertyPath::ZeroOrMore(p) => {
3246 let start2 = start.clone();
3247 let eval = self.clone();
3248 let p = Rc::clone(p);
3249 self.run_if_term_is_a_dataset_node(start, move |graph_name| {
3250 let eval = eval.clone();
3251 let p = Rc::clone(&p);
3252 let graph_name2 = graph_name.clone();
3253 transitive_closure(Some(Ok(start2.clone())), move |e| {
3254 eval.eval_from_in_graph(&p, &e, graph_name2.as_ref())
3255 })
3256 .map(move |e| Ok((e?, graph_name.clone())))
3257 })
3258 }
3259 PropertyPath::OneOrMore(p) => {
3260 let eval = self.clone();
3261 let p = Rc::clone(p);
3262 Box::new(transitive_closure(
3263 self.eval_from_in_unknown_graph(&p, start),
3264 move |(e, graph_name)| {
3265 eval.eval_from_in_graph(&p, &e, graph_name.as_ref())
3266 .map(move |e| Ok((e?, graph_name.clone())))
3267 },
3268 ))
3269 }
3270 PropertyPath::ZeroOrOne(p) => {
3271 let eval = self.clone();
3272 let start2 = start.clone();
3273 let p = Rc::clone(p);
3274 self.run_if_term_is_a_dataset_node(start, move |graph_name| {
3275 hash_deduplicate(once(Ok(start2.clone())).chain(eval.eval_from_in_graph(
3276 &p,
3277 &start2,
3278 graph_name.as_ref(),
3279 )))
3280 .map(move |e| Ok((e?, graph_name.clone())))
3281 })
3282 }
3283 PropertyPath::NegatedPropertySet(ps) => {
3284 let ps = Rc::clone(ps);
3285 Box::new(
3286 self.dataset
3287 .internal_quads_for_pattern(Some(start), None, None, None)
3288 .filter_map(move |t| match t {
3289 Ok(t) => {
3290 if ps.contains(&t.predicate) {
3291 None
3292 } else {
3293 Some(Ok((t.object, t.graph_name)))
3294 }
3295 }
3296 Err(e) => Some(Err(e)),
3297 }),
3298 )
3299 }
3300 }
3301 }
3302
3303 fn eval_to_in_graph(
3304 &self,
3305 path: &PropertyPath<D::InternalTerm>,
3306 end: &D::InternalTerm,
3307 graph_name: Option<&D::InternalTerm>,
3308 ) -> Box<dyn Iterator<Item = Result<D::InternalTerm, QueryEvaluationError>> + 'a> {
3309 match path {
3310 PropertyPath::Path(p) => Box::new(
3311 self.dataset
3312 .internal_quads_for_pattern(None, Some(p), Some(end), Some(graph_name))
3313 .map(|t| Ok(t?.subject)),
3314 ),
3315 PropertyPath::Reverse(p) => self.eval_from_in_graph(p, end, graph_name),
3316 PropertyPath::Sequence(a, b) => {
3317 let eval = self.clone();
3318 let a = Rc::clone(a);
3319 let graph_name2 = graph_name.cloned();
3320 Box::new(
3321 self.eval_to_in_graph(b, end, graph_name)
3322 .flat_map_ok(move |middle| {
3323 eval.eval_to_in_graph(&a, &middle, graph_name2.as_ref())
3324 }),
3325 )
3326 }
3327 PropertyPath::Alternative(a, b) => Box::new(hash_deduplicate(
3328 self.eval_to_in_graph(a, end, graph_name)
3329 .chain(self.eval_to_in_graph(b, end, graph_name)),
3330 )),
3331 PropertyPath::ZeroOrMore(p) => {
3332 self.run_if_term_is_a_graph_node(end, graph_name, || {
3333 let eval = self.clone();
3334 let p = Rc::clone(p);
3335 let graph_name2 = graph_name.cloned();
3336 transitive_closure(Some(Ok(end.clone())), move |e| {
3337 eval.eval_to_in_graph(&p, &e, graph_name2.as_ref())
3338 })
3339 })
3340 }
3341 PropertyPath::OneOrMore(p) => {
3342 let eval = self.clone();
3343 let p = Rc::clone(p);
3344 let graph_name2 = graph_name.cloned();
3345 Box::new(transitive_closure(
3346 self.eval_to_in_graph(&p, end, graph_name),
3347 move |e| eval.eval_to_in_graph(&p, &e, graph_name2.as_ref()),
3348 ))
3349 }
3350 PropertyPath::ZeroOrOne(p) => self.run_if_term_is_a_graph_node(end, graph_name, || {
3351 hash_deduplicate(
3352 once(Ok(end.clone())).chain(self.eval_to_in_graph(p, end, graph_name)),
3353 )
3354 }),
3355 PropertyPath::NegatedPropertySet(ps) => {
3356 let ps = Rc::clone(ps);
3357 Box::new(
3358 self.dataset
3359 .internal_quads_for_pattern(None, None, Some(end), Some(graph_name))
3360 .filter_map(move |t| match t {
3361 Ok(t) => {
3362 if ps.contains(&t.predicate) {
3363 None
3364 } else {
3365 Some(Ok(t.subject))
3366 }
3367 }
3368 Err(e) => Some(Err(e)),
3369 }),
3370 )
3371 }
3372 }
3373 }
3374
3375 fn eval_to_in_unknown_graph(
3376 &self,
3377 path: &PropertyPath<D::InternalTerm>,
3378 end: &D::InternalTerm,
3379 ) -> Box<
3380 dyn Iterator<
3381 Item = Result<(D::InternalTerm, Option<D::InternalTerm>), QueryEvaluationError>,
3382 > + 'a,
3383 > {
3384 match path {
3385 PropertyPath::Path(p) => Box::new(
3386 self.dataset
3387 .internal_quads_for_pattern(None, Some(p), Some(end), None)
3388 .map(|t| {
3389 let t = t?;
3390 Ok((t.subject, t.graph_name))
3391 }),
3392 ),
3393 PropertyPath::Reverse(p) => self.eval_from_in_unknown_graph(p, end),
3394 PropertyPath::Sequence(a, b) => {
3395 let eval = self.clone();
3396 let a = Rc::clone(a);
3397 Box::new(self.eval_to_in_unknown_graph(b, end).flat_map_ok(
3398 move |(middle, graph_name)| {
3399 eval.eval_to_in_graph(&a, &middle, graph_name.as_ref())
3400 .map(move |start| Ok((start?, graph_name.clone())))
3401 },
3402 ))
3403 }
3404 PropertyPath::Alternative(a, b) => Box::new(hash_deduplicate(
3405 self.eval_to_in_unknown_graph(a, end)
3406 .chain(self.eval_to_in_unknown_graph(b, end)),
3407 )),
3408 PropertyPath::ZeroOrMore(p) => {
3409 let end2 = end.clone();
3410 let eval = self.clone();
3411 let p = Rc::clone(p);
3412 self.run_if_term_is_a_dataset_node(end, move |graph_name| {
3413 let eval = eval.clone();
3414 let p = Rc::clone(&p);
3415 let graph_name2 = graph_name.clone();
3416 transitive_closure(Some(Ok(end2.clone())), move |e| {
3417 eval.eval_to_in_graph(&p, &e, graph_name2.as_ref())
3418 })
3419 .map(move |e| Ok((e?, graph_name.clone())))
3420 })
3421 }
3422 PropertyPath::OneOrMore(p) => {
3423 let eval = self.clone();
3424 let p = Rc::clone(p);
3425 Box::new(transitive_closure(
3426 self.eval_to_in_unknown_graph(&p, end),
3427 move |(e, graph_name)| {
3428 eval.eval_to_in_graph(&p, &e, graph_name.as_ref())
3429 .map(move |e| Ok((e?, graph_name.clone())))
3430 },
3431 ))
3432 }
3433 PropertyPath::ZeroOrOne(p) => {
3434 let eval = self.clone();
3435 let end2 = end.clone();
3436 let p = Rc::clone(p);
3437 self.run_if_term_is_a_dataset_node(end, move |graph_name| {
3438 hash_deduplicate(once(Ok(end2.clone())).chain(eval.eval_to_in_graph(
3439 &p,
3440 &end2,
3441 graph_name.as_ref(),
3442 )))
3443 .map(move |e| Ok((e?, graph_name.clone())))
3444 })
3445 }
3446 PropertyPath::NegatedPropertySet(ps) => {
3447 let ps = Rc::clone(ps);
3448 Box::new(
3449 self.dataset
3450 .internal_quads_for_pattern(None, None, Some(end), None)
3451 .filter_map(move |t| match t {
3452 Ok(t) => {
3453 if ps.contains(&t.predicate) {
3454 None
3455 } else {
3456 Some(Ok((t.subject, t.graph_name)))
3457 }
3458 }
3459 Err(e) => Some(Err(e)),
3460 }),
3461 )
3462 }
3463 }
3464 }
3465
3466 fn eval_open_in_graph(
3467 &self,
3468 path: &PropertyPath<D::InternalTerm>,
3469 graph_name: Option<&D::InternalTerm>,
3470 ) -> Box<
3471 dyn Iterator<Item = Result<(D::InternalTerm, D::InternalTerm), QueryEvaluationError>> + 'a,
3472 > {
3473 match path {
3474 PropertyPath::Path(p) => Box::new(
3475 self.dataset
3476 .internal_quads_for_pattern(None, Some(p), None, Some(graph_name))
3477 .map(|t| {
3478 let t = t?;
3479 Ok((t.subject, t.object))
3480 }),
3481 ),
3482 PropertyPath::Reverse(p) => Box::new(
3483 self.eval_open_in_graph(p, graph_name)
3484 .map(|t| t.map(|(s, o)| (o, s))),
3485 ),
3486 PropertyPath::Sequence(a, b) => {
3487 let eval = self.clone();
3488 let b = Rc::clone(b);
3489 let graph_name2 = graph_name.cloned();
3490 Box::new(self.eval_open_in_graph(a, graph_name).flat_map_ok(
3491 move |(start, middle)| {
3492 eval.eval_from_in_graph(&b, &middle, graph_name2.as_ref())
3493 .map(move |end| Ok((start.clone(), end?)))
3494 },
3495 ))
3496 }
3497 PropertyPath::Alternative(a, b) => Box::new(hash_deduplicate(
3498 self.eval_open_in_graph(a, graph_name)
3499 .chain(self.eval_open_in_graph(b, graph_name)),
3500 )),
3501 PropertyPath::ZeroOrMore(p) => {
3502 let eval = self.clone();
3503 let p = Rc::clone(p);
3504 let graph_name2 = graph_name.cloned();
3505 Box::new(transitive_closure(
3506 self.get_subject_or_object_identity_pairs_in_graph(graph_name),
3507 move |(start, middle)| {
3508 eval.eval_from_in_graph(&p, &middle, graph_name2.as_ref())
3509 .map(move |end| Ok((start.clone(), end?)))
3510 },
3511 ))
3512 }
3513 PropertyPath::OneOrMore(p) => {
3514 let eval = self.clone();
3515 let p = Rc::clone(p);
3516 let graph_name2 = graph_name.cloned();
3517 Box::new(transitive_closure(
3518 self.eval_open_in_graph(&p, graph_name),
3519 move |(start, middle)| {
3520 eval.eval_from_in_graph(&p, &middle, graph_name2.as_ref())
3521 .map(move |end| Ok((start.clone(), end?)))
3522 },
3523 ))
3524 }
3525 PropertyPath::ZeroOrOne(p) => Box::new(hash_deduplicate(
3526 self.get_subject_or_object_identity_pairs_in_graph(graph_name)
3527 .chain(self.eval_open_in_graph(p, graph_name)),
3528 )),
3529 PropertyPath::NegatedPropertySet(ps) => {
3530 let ps = Rc::clone(ps);
3531 Box::new(
3532 self.dataset
3533 .internal_quads_for_pattern(None, None, None, Some(graph_name))
3534 .filter_map(move |t| match t {
3535 Ok(t) => {
3536 if ps.contains(&t.predicate) {
3537 None
3538 } else {
3539 Some(Ok((t.subject, t.object)))
3540 }
3541 }
3542 Err(e) => Some(Err(e)),
3543 }),
3544 )
3545 }
3546 }
3547 }
3548
3549 fn eval_open_in_unknown_graph(
3550 &self,
3551 path: &PropertyPath<D::InternalTerm>,
3552 ) -> Box<
3553 dyn Iterator<
3554 Item = Result<
3555 (D::InternalTerm, D::InternalTerm, Option<D::InternalTerm>),
3556 QueryEvaluationError,
3557 >,
3558 > + 'a,
3559 > {
3560 match path {
3561 PropertyPath::Path(p) => Box::new(
3562 self.dataset
3563 .internal_quads_for_pattern(None, Some(p), None, None)
3564 .map(|t| {
3565 let t = t?;
3566 Ok((t.subject, t.object, t.graph_name))
3567 }),
3568 ),
3569 PropertyPath::Reverse(p) => Box::new(
3570 self.eval_open_in_unknown_graph(p)
3571 .map(|t| t.map(|(s, o, g)| (o, s, g))),
3572 ),
3573 PropertyPath::Sequence(a, b) => {
3574 let eval = self.clone();
3575 let b = Rc::clone(b);
3576 Box::new(self.eval_open_in_unknown_graph(a).flat_map_ok(
3577 move |(start, middle, graph_name)| {
3578 eval.eval_from_in_graph(&b, &middle, graph_name.as_ref())
3579 .map(move |end| Ok((start.clone(), end?, graph_name.clone())))
3580 },
3581 ))
3582 }
3583 PropertyPath::Alternative(a, b) => Box::new(hash_deduplicate(
3584 self.eval_open_in_unknown_graph(a)
3585 .chain(self.eval_open_in_unknown_graph(b)),
3586 )),
3587 PropertyPath::ZeroOrMore(p) => {
3588 let eval = self.clone();
3589 let p = Rc::clone(p);
3590 Box::new(transitive_closure(
3591 self.get_subject_or_object_identity_pairs_in_dataset(),
3592 move |(start, middle, graph_name)| {
3593 eval.eval_from_in_graph(&p, &middle, graph_name.as_ref())
3594 .map(move |end| Ok((start.clone(), end?, graph_name.clone())))
3595 },
3596 ))
3597 }
3598 PropertyPath::OneOrMore(p) => {
3599 let eval = self.clone();
3600 let p = Rc::clone(p);
3601 Box::new(transitive_closure(
3602 self.eval_open_in_unknown_graph(&p),
3603 move |(start, middle, graph_name)| {
3604 eval.eval_from_in_graph(&p, &middle, graph_name.as_ref())
3605 .map(move |end| Ok((start.clone(), end?, graph_name.clone())))
3606 },
3607 ))
3608 }
3609 PropertyPath::ZeroOrOne(p) => Box::new(hash_deduplicate(
3610 self.get_subject_or_object_identity_pairs_in_dataset()
3611 .chain(self.eval_open_in_unknown_graph(p)),
3612 )),
3613 PropertyPath::NegatedPropertySet(ps) => {
3614 let ps = Rc::clone(ps);
3615 Box::new(
3616 self.dataset
3617 .internal_quads_for_pattern(None, None, None, None)
3618 .filter_map(move |t| match t {
3619 Ok(t) => {
3620 if ps.contains(&t.predicate) {
3621 None
3622 } else {
3623 Some(Ok((t.subject, t.object, t.graph_name)))
3624 }
3625 }
3626 Err(e) => Some(Err(e)),
3627 }),
3628 )
3629 }
3630 }
3631 }
3632
3633 fn get_subject_or_object_identity_pairs_in_graph(
3634 &self,
3635 graph_name: Option<&D::InternalTerm>,
3636 ) -> impl Iterator<Item = Result<(D::InternalTerm, D::InternalTerm), QueryEvaluationError>>
3637 + use<'a, D> {
3638 self.dataset
3639 .internal_quads_for_pattern(None, None, None, Some(graph_name))
3640 .flat_map_ok(|t| {
3641 [
3642 Ok((t.subject.clone(), t.subject)),
3643 Ok((t.object.clone(), t.object)),
3644 ]
3645 })
3646 }
3647
3648 fn get_subject_or_object_identity_pairs_in_dataset(
3649 &self,
3650 ) -> impl Iterator<
3651 Item = Result<
3652 (D::InternalTerm, D::InternalTerm, Option<D::InternalTerm>),
3653 QueryEvaluationError,
3654 >,
3655 > + use<'a, D> {
3656 self.dataset
3657 .internal_quads_for_pattern(None, None, None, None)
3658 .flat_map_ok(|t| {
3659 [
3660 Ok((t.subject.clone(), t.subject, t.graph_name.clone())),
3661 Ok((t.object.clone(), t.object, t.graph_name)),
3662 ]
3663 })
3664 }
3665
3666 fn run_if_term_is_a_graph_node<
3667 T: 'a,
3668 I: Iterator<Item = Result<T, QueryEvaluationError>> + 'a,
3669 >(
3670 &self,
3671 term: &D::InternalTerm,
3672 graph_name: Option<&D::InternalTerm>,
3673 f: impl FnOnce() -> I,
3674 ) -> Box<dyn Iterator<Item = Result<T, QueryEvaluationError>> + 'a> {
3675 match self.is_subject_or_object_in_graph(term, graph_name) {
3676 Ok(true) => Box::new(f()),
3677 Ok(false) => {
3678 Box::new(empty()) }
3680 Err(error) => Box::new(once(Err(error))),
3681 }
3682 }
3683
3684 fn is_subject_or_object_in_graph(
3685 &self,
3686 term: &D::InternalTerm,
3687 graph_name: Option<&D::InternalTerm>,
3688 ) -> Result<bool, QueryEvaluationError> {
3689 Ok(self
3690 .dataset
3691 .internal_quads_for_pattern(Some(term), None, None, Some(graph_name))
3692 .next()
3693 .transpose()?
3694 .is_some()
3695 || self
3696 .dataset
3697 .internal_quads_for_pattern(None, None, Some(term), Some(graph_name))
3698 .next()
3699 .transpose()?
3700 .is_some())
3701 }
3702
3703 fn run_if_term_is_a_dataset_node<
3704 T: 'a,
3705 I: IntoIterator<Item = Result<T, QueryEvaluationError>> + 'a,
3706 >(
3707 &self,
3708 term: &D::InternalTerm,
3709 f: impl FnMut(Option<D::InternalTerm>) -> I + 'a,
3710 ) -> Box<dyn Iterator<Item = Result<T, QueryEvaluationError>> + 'a> {
3711 match self
3712 .find_graphs_where_the_node_is_in(term)
3713 .collect::<Result<FxHashSet<_>, _>>()
3714 {
3715 Ok(graph_names) => Box::new(graph_names.into_iter().flat_map(f)),
3716 Err(error) => Box::new(once(Err(error))),
3717 }
3718 }
3719
3720 fn find_graphs_where_the_node_is_in(
3721 &self,
3722 term: &D::InternalTerm,
3723 ) -> impl Iterator<Item = Result<Option<D::InternalTerm>, QueryEvaluationError>> + use<'a, D>
3724 {
3725 self.dataset
3726 .internal_quads_for_pattern(Some(term), None, None, None)
3727 .chain(
3728 self.dataset
3729 .internal_quads_for_pattern(None, None, Some(term), None),
3730 )
3731 .map(|q| Ok(q?.graph_name))
3732 }
3733}
3734
3735impl<'a, D: QueryableDataset<'a>> Clone for PathEvaluator<'a, D> {
3736 fn clone(&self) -> Self {
3737 Self {
3738 dataset: self.dataset.clone(),
3739 }
3740 }
3741}
3742
3743struct CartesianProductJoinIterator<'a, T> {
3744 probe_iter: Peekable<InternalTuplesIterator<'a, T>>,
3745 built: Vec<InternalTuple<T>>,
3746 buffered_results: Vec<Result<InternalTuple<T>, QueryEvaluationError>>,
3747}
3748
3749impl<T: Clone + Eq> Iterator for CartesianProductJoinIterator<'_, T> {
3750 type Item = Result<InternalTuple<T>, QueryEvaluationError>;
3751
3752 fn next(&mut self) -> Option<Self::Item> {
3753 loop {
3754 if let Some(result) = self.buffered_results.pop() {
3755 return Some(result);
3756 }
3757 let probe_tuple = match self.probe_iter.next()? {
3758 Ok(probe_tuple) => probe_tuple,
3759 Err(error) => return Some(Err(error)),
3760 };
3761 for built_tuple in &self.built {
3762 if let Some(result_tuple) = probe_tuple.combine_with(built_tuple) {
3763 self.buffered_results.push(Ok(result_tuple))
3764 }
3765 }
3766 }
3767 }
3768
3769 fn size_hint(&self) -> (usize, Option<usize>) {
3770 let (min, max) = self.probe_iter.size_hint();
3771 (
3772 min.saturating_mul(self.built.len()),
3773 max.map(|v| v.saturating_mul(self.built.len())),
3774 )
3775 }
3776}
3777
3778struct HashJoinIterator<'a, T> {
3779 probe_iter: Peekable<InternalTuplesIterator<'a, T>>,
3780 built: InternalTupleSet<T>,
3781 buffered_results: Vec<Result<InternalTuple<T>, QueryEvaluationError>>,
3782}
3783
3784impl<T: Clone + Eq + Hash> Iterator for HashJoinIterator<'_, T> {
3785 type Item = Result<InternalTuple<T>, QueryEvaluationError>;
3786
3787 fn next(&mut self) -> Option<Self::Item> {
3788 loop {
3789 if let Some(result) = self.buffered_results.pop() {
3790 return Some(result);
3791 }
3792 let probe_tuple = match self.probe_iter.next()? {
3793 Ok(probe_tuple) => probe_tuple,
3794 Err(error) => return Some(Err(error)),
3795 };
3796 self.buffered_results.extend(
3797 self.built
3798 .get(&probe_tuple)
3799 .iter()
3800 .filter_map(|built_tuple| probe_tuple.combine_with(built_tuple).map(Ok)),
3801 )
3802 }
3803 }
3804
3805 fn size_hint(&self) -> (usize, Option<usize>) {
3806 (
3807 0,
3808 self.probe_iter
3809 .size_hint()
3810 .1
3811 .map(|v| v.saturating_mul(self.built.len())),
3812 )
3813 }
3814}
3815
3816struct HashLeftJoinIterator<'a, T> {
3817 left_iter: InternalTuplesIterator<'a, T>,
3818 right: InternalTupleSet<T>,
3819 buffered_results: Vec<Result<InternalTuple<T>, QueryEvaluationError>>,
3820 expression: Rc<dyn Fn(&InternalTuple<T>) -> Option<bool> + 'a>,
3821}
3822
3823impl<T: Clone + Eq + Hash> Iterator for HashLeftJoinIterator<'_, T> {
3824 type Item = Result<InternalTuple<T>, QueryEvaluationError>;
3825
3826 fn next(&mut self) -> Option<Self::Item> {
3827 loop {
3828 if let Some(result) = self.buffered_results.pop() {
3829 return Some(result);
3830 }
3831 let left_tuple = match self.left_iter.next()? {
3832 Ok(left_tuple) => left_tuple,
3833 Err(error) => return Some(Err(error)),
3834 };
3835 self.buffered_results.extend(
3836 self.right
3837 .get(&left_tuple)
3838 .iter()
3839 .filter_map(|right_tuple| left_tuple.combine_with(right_tuple))
3840 .filter(|tuple| (self.expression)(tuple).unwrap_or(false))
3841 .map(Ok),
3842 );
3843 if self.buffered_results.is_empty() {
3844 return Some(Ok(left_tuple));
3846 }
3847 }
3848 }
3849
3850 fn size_hint(&self) -> (usize, Option<usize>) {
3851 (
3852 0,
3853 self.left_iter
3854 .size_hint()
3855 .1
3856 .map(|v| v.saturating_mul(self.right.len())),
3857 )
3858 }
3859}
3860
3861#[cfg(feature = "sep-0006")]
3862struct ForLoopLeftJoinIterator<'a, T> {
3863 right_evaluator: InternalTupleEvaluator<'a, T>,
3864 left_iter: InternalTuplesIterator<'a, T>,
3865 current_right: InternalTuplesIterator<'a, T>,
3866 left_tuple_to_yield: Option<InternalTuple<T>>,
3867}
3868
3869#[cfg(feature = "sep-0006")]
3870impl<T: Clone> Iterator for ForLoopLeftJoinIterator<'_, T> {
3871 type Item = Result<InternalTuple<T>, QueryEvaluationError>;
3872
3873 fn next(&mut self) -> Option<Self::Item> {
3874 loop {
3875 if let Some(tuple) = self.current_right.next() {
3876 if tuple.is_ok() {
3877 self.left_tuple_to_yield = None;
3879 }
3880 return Some(tuple);
3881 }
3882 if let Some(left_tuple) = self.left_tuple_to_yield.take() {
3883 return Some(Ok(left_tuple));
3884 }
3885 let left_tuple = match self.left_iter.next()? {
3886 Ok(left_tuple) => left_tuple,
3887 Err(error) => return Some(Err(error)),
3888 };
3889 self.current_right = (self.right_evaluator)(left_tuple.clone());
3890 self.left_tuple_to_yield = Some(left_tuple);
3891 }
3892 }
3893}
3894
3895struct UnionIterator<'a, T> {
3896 plans: Vec<Rc<dyn Fn(InternalTuple<T>) -> InternalTuplesIterator<'a, T> + 'a>>,
3897 input: InternalTuple<T>,
3898 current_iterator: InternalTuplesIterator<'a, T>,
3899 current_plan: usize,
3900}
3901
3902impl<T: Clone> Iterator for UnionIterator<'_, T> {
3903 type Item = Result<InternalTuple<T>, QueryEvaluationError>;
3904
3905 fn next(&mut self) -> Option<Self::Item> {
3906 loop {
3907 if let Some(tuple) = self.current_iterator.next() {
3908 return Some(tuple);
3909 }
3910 if self.current_plan >= self.plans.len() {
3911 return None;
3912 }
3913 self.current_iterator = self.plans[self.current_plan](self.input.clone());
3914 self.current_plan += 1;
3915 }
3916 }
3917}
3918
3919struct ConsecutiveDeduplication<'a, T> {
3920 inner: InternalTuplesIterator<'a, T>,
3921 current: Option<InternalTuple<T>>,
3922}
3923
3924impl<T: Eq> Iterator for ConsecutiveDeduplication<'_, T> {
3925 type Item = Result<InternalTuple<T>, QueryEvaluationError>;
3926
3927 fn next(&mut self) -> Option<Self::Item> {
3928 loop {
3930 if let Some(next) = self.inner.next() {
3931 match next {
3932 Ok(next) => match self.current.take() {
3933 Some(current) if current != next => {
3934 self.current = Some(next);
3936 return Some(Ok(current));
3937 }
3938 _ => {
3939 self.current = Some(next);
3941 }
3942 },
3943 Err(error) => return Some(Err(error)), }
3945 } else {
3946 return self.current.take().map(Ok);
3947 }
3948 }
3949 }
3950
3951 fn size_hint(&self) -> (usize, Option<usize>) {
3952 let (min, max) = self.inner.size_hint();
3953 ((min != 0).into(), max)
3954 }
3955}
3956
3957struct ConstructIterator<'a, D: QueryableDataset<'a>> {
3958 eval: SimpleEvaluator<'a, D>,
3959 iter: InternalTuplesIterator<'a, D::InternalTerm>,
3960 template: Vec<TripleTemplate>,
3961 buffered_results: Vec<Result<Triple, QueryEvaluationError>>,
3962 already_emitted_results: FxHashSet<Triple>,
3963 bnodes: Vec<BlankNode>,
3964}
3965
3966impl<'a, D: QueryableDataset<'a>> Iterator for ConstructIterator<'a, D> {
3967 type Item = Result<Triple, QueryEvaluationError>;
3968
3969 fn next(&mut self) -> Option<Self::Item> {
3970 loop {
3971 if let Some(result) = self.buffered_results.pop() {
3972 return Some(result);
3973 }
3974 {
3975 let tuple = match self.iter.next()? {
3976 Ok(tuple) => tuple,
3977 Err(error) => return Some(Err(error)),
3978 };
3979 for template in &self.template {
3980 if let (Some(subject), Some(predicate), Some(object)) = (
3981 get_triple_template_value(
3982 &template.subject,
3983 &tuple,
3984 &mut self.bnodes,
3985 &self.eval.dataset,
3986 )
3987 .and_then(|t| t.try_into().ok()),
3988 get_triple_template_value(
3989 &template.predicate,
3990 &tuple,
3991 &mut self.bnodes,
3992 &self.eval.dataset,
3993 )
3994 .and_then(|t| t.try_into().ok()),
3995 get_triple_template_value(
3996 &template.object,
3997 &tuple,
3998 &mut self.bnodes,
3999 &self.eval.dataset,
4000 ),
4001 ) {
4002 let triple = Triple {
4003 subject,
4004 predicate,
4005 object,
4006 };
4007 #[cfg(feature = "sparql-12")]
4010 let new_triple = triple.subject.is_blank_node()
4011 || triple.object.is_blank_node()
4012 || triple.object.is_triple()
4013 || self.already_emitted_results.insert(triple.clone());
4014 #[cfg(not(feature = "sparql-12"))]
4015 let new_triple = triple.subject.is_blank_node()
4016 || triple.object.is_blank_node()
4017 || self.already_emitted_results.insert(triple.clone());
4018 if new_triple {
4019 self.buffered_results.push(Ok(triple));
4020 if self.already_emitted_results.len() > 1024 * 1024 {
4021 self.already_emitted_results.clear();
4023 }
4024 }
4025 }
4026 }
4027 self.bnodes.clear(); }
4029 }
4030 }
4031
4032 fn size_hint(&self) -> (usize, Option<usize>) {
4033 let (min, max) = self.iter.size_hint();
4034 (
4035 min.saturating_mul(self.template.len()),
4036 max.map(|v| v.saturating_mul(self.template.len())),
4037 )
4038 }
4039}
4040
4041pub struct TripleTemplate {
4042 pub subject: TripleTemplateValue,
4043 pub predicate: TripleTemplateValue,
4044 pub object: TripleTemplateValue,
4045}
4046
4047pub enum TripleTemplateValue {
4048 Constant(Term),
4049 BlankNode(usize),
4050 Variable(usize),
4051 #[cfg(feature = "sparql-12")]
4052 Triple(Box<TripleTemplate>),
4053}
4054
4055impl TripleTemplateValue {
4056 #[cfg_attr(not(feature = "sparql-12"), expect(clippy::unnecessary_wraps))]
4057 fn from_term_or_variable(
4058 term_or_variable: &TermPattern,
4059 variables: &mut Vec<Variable>,
4060 bnodes: &mut Vec<BlankNode>,
4061 ) -> Option<Self> {
4062 Some(match term_or_variable {
4063 TermPattern::Variable(variable) => Self::Variable(encode_variable(variables, variable)),
4064 TermPattern::NamedNode(node) => Self::Constant(node.clone().into()),
4065 TermPattern::BlankNode(bnode) => Self::BlankNode(bnode_key(bnodes, bnode)),
4066 TermPattern::Literal(literal) => Self::Constant(literal.clone().into()),
4067 #[cfg(feature = "sparql-12")]
4068 TermPattern::Triple(triple) => {
4069 match (
4070 Self::from_term_or_variable(&triple.subject, variables, bnodes)?,
4071 Self::from_named_node_or_variable(&triple.predicate, variables),
4072 Self::from_term_or_variable(&triple.object, variables, bnodes)?,
4073 ) {
4074 (
4075 Self::Constant(subject),
4076 Self::Constant(predicate),
4077 Self::Constant(object),
4078 ) => Self::Constant(
4079 Triple {
4080 subject: subject.try_into().ok()?,
4081 predicate: predicate.try_into().ok()?,
4082 object,
4083 }
4084 .into(),
4085 ),
4086 (subject, predicate, object) => {
4087 TripleTemplateValue::Triple(Box::new(TripleTemplate {
4088 subject,
4089 predicate,
4090 object,
4091 }))
4092 }
4093 }
4094 }
4095 })
4096 }
4097
4098 fn from_named_node_or_variable(
4099 named_node_or_variable: &NamedNodePattern,
4100 variables: &mut Vec<Variable>,
4101 ) -> TripleTemplateValue {
4102 match named_node_or_variable {
4103 NamedNodePattern::Variable(variable) => {
4104 Self::Variable(encode_variable(variables, variable))
4105 }
4106 NamedNodePattern::NamedNode(term) => Self::Constant(term.clone().into()),
4107 }
4108 }
4109}
4110
4111fn get_triple_template_value<'a, D: QueryableDataset<'a>>(
4112 selector: &TripleTemplateValue,
4113 tuple: &InternalTuple<D::InternalTerm>,
4114 bnodes: &mut Vec<BlankNode>,
4115 dataset: &EvalDataset<'a, D>,
4116) -> Option<Term> {
4117 match selector {
4118 TripleTemplateValue::Constant(term) => Some(term.clone()),
4119 TripleTemplateValue::Variable(v) => {
4120 let t = tuple.get(*v)?;
4121 dataset.externalize_term(t.clone()).ok() }
4123 TripleTemplateValue::BlankNode(bnode) => {
4124 if *bnode >= bnodes.len() {
4125 bnodes.resize_with(*bnode + 1, BlankNode::default)
4126 }
4127 Some(bnodes[*bnode].clone().into())
4128 }
4129 #[cfg(feature = "sparql-12")]
4130 TripleTemplateValue::Triple(triple) => Some(
4131 Triple {
4132 subject: get_triple_template_value(&triple.subject, tuple, bnodes, dataset)?
4133 .try_into()
4134 .ok()?,
4135 predicate: get_triple_template_value(&triple.predicate, tuple, bnodes, dataset)?
4136 .try_into()
4137 .ok()?,
4138 object: get_triple_template_value(&triple.object, tuple, bnodes, dataset)?,
4139 }
4140 .into(),
4141 ),
4142 }
4143}
4144
4145struct DescribeIterator<'a, D: QueryableDataset<'a>> {
4146 eval: SimpleEvaluator<'a, D>,
4147 tuples_to_describe: InternalTuplesIterator<'a, D::InternalTerm>,
4148 nodes_described: FxHashSet<D::InternalTerm>,
4149 nodes_to_describe: Vec<D::InternalTerm>,
4150 quads:
4151 Box<dyn Iterator<Item = Result<InternalQuad<D::InternalTerm>, QueryEvaluationError>> + 'a>,
4152}
4153
4154impl<'a, D: QueryableDataset<'a>> Iterator for DescribeIterator<'a, D> {
4155 type Item = Result<Triple, QueryEvaluationError>;
4156
4157 fn next(&mut self) -> Option<Self::Item> {
4158 loop {
4159 if let Some(quad) = self.quads.next() {
4160 let quad = match quad {
4161 Ok(quad) => quad,
4162 Err(error) => return Some(Err(error)),
4163 };
4164 let subject = match self.eval.dataset.externalize_term(quad.subject) {
4166 Ok(t) => t,
4167 Err(e) => return Some(Err(e)),
4168 };
4169 let predicate = match self.eval.dataset.externalize_term(quad.predicate) {
4170 Ok(t) => t,
4171 Err(e) => return Some(Err(e)),
4172 };
4173 let object = match self.eval.dataset.externalize_term(quad.object.clone()) {
4174 Ok(t) => t,
4175 Err(e) => return Some(Err(e)),
4176 };
4177 if object.is_blank_node() && self.nodes_described.insert(quad.object.clone()) {
4179 self.nodes_to_describe.push(quad.object);
4180 }
4181 return Some(Ok(Triple {
4182 subject: subject.try_into().ok()?,
4183 predicate: predicate.try_into().ok()?,
4184 object,
4185 }));
4186 }
4187 if let Some(node_to_describe) = self.nodes_to_describe.pop() {
4188 self.quads = self.eval.dataset.internal_quads_for_pattern(
4190 Some(&node_to_describe),
4191 None,
4192 None,
4193 Some(None),
4194 );
4195 } else {
4196 let tuple = match self.tuples_to_describe.next()? {
4197 Ok(tuple) => tuple,
4198 Err(error) => return Some(Err(error)),
4199 };
4200 for node in tuple.into_iter().flatten() {
4201 if self.nodes_described.insert(node.clone()) {
4202 self.nodes_to_describe.push(node);
4203 }
4204 }
4205 }
4206 }
4207 }
4208}
4209
4210fn transitive_closure<T: Clone + Eq + Hash, E, NI: Iterator<Item = Result<T, E>>>(
4211 start: impl IntoIterator<Item = Result<T, E>>,
4212 mut next: impl FnMut(T) -> NI,
4213) -> impl Iterator<Item = Result<T, E>> {
4214 let mut errors = Vec::new();
4215 let mut todo = start
4216 .into_iter()
4217 .filter_map(|e| match e {
4218 Ok(e) => Some(e),
4219 Err(e) => {
4220 errors.push(e);
4221 None
4222 }
4223 })
4224 .collect::<Vec<_>>();
4225 let mut all = todo.iter().cloned().collect::<FxHashSet<_>>();
4226 while let Some(e) = todo.pop() {
4227 for e in next(e) {
4228 match e {
4229 Ok(e) => {
4230 if all.insert(e.clone()) {
4231 todo.push(e)
4232 }
4233 }
4234 Err(e) => errors.push(e),
4235 }
4236 }
4237 }
4238 errors.into_iter().map(Err).chain(all.into_iter().map(Ok))
4239}
4240
4241fn look_in_transitive_closure<T: Clone + Eq + Hash, E, NI: Iterator<Item = Result<T, E>>>(
4242 start: impl IntoIterator<Item = Result<T, E>>,
4243 mut next: impl FnMut(T) -> NI,
4244 target: &T,
4245) -> Result<bool, E> {
4246 let mut todo = start.into_iter().collect::<Result<Vec<_>, _>>()?;
4247 let mut all = todo.iter().cloned().collect::<FxHashSet<_>>();
4248 while let Some(e) = todo.pop() {
4249 if e == *target {
4250 return Ok(true);
4251 }
4252 for e in next(e) {
4253 let e = e?;
4254 if all.insert(e.clone()) {
4255 todo.push(e);
4256 }
4257 }
4258 }
4259 Ok(false)
4260}
4261
4262fn hash_deduplicate<T: Eq + Hash + Clone, E>(
4263 iter: impl Iterator<Item = Result<T, E>>,
4264) -> impl Iterator<Item = Result<T, E>> {
4265 let mut already_seen = FxHashSet::with_capacity_and_hasher(iter.size_hint().0, FxBuildHasher);
4266 iter.filter(move |e| {
4267 if let Ok(e) = e {
4268 if already_seen.contains(e) {
4269 false
4270 } else {
4271 already_seen.insert(e.clone());
4272 true
4273 }
4274 } else {
4275 true
4276 }
4277 })
4278}
4279
4280trait ResultIterator<T, E>: Iterator<Item = Result<T, E>> + Sized {
4281 fn flat_map_ok<O, F: FnMut(T) -> U, U: IntoIterator<Item = Result<O, E>>>(
4282 self,
4283 f: F,
4284 ) -> FlatMapOk<T, E, O, Self, F, U>;
4285}
4286
4287impl<T, E, I: Iterator<Item = Result<T, E>> + Sized> ResultIterator<T, E> for I {
4288 #[inline]
4289 fn flat_map_ok<O, F: FnMut(T) -> U, U: IntoIterator<Item = Result<O, E>>>(
4290 self,
4291 f: F,
4292 ) -> FlatMapOk<T, E, O, Self, F, U> {
4293 FlatMapOk {
4294 inner: self,
4295 f,
4296 current: None,
4297 }
4298 }
4299}
4300
4301struct FlatMapOk<
4302 T,
4303 E,
4304 O,
4305 I: Iterator<Item = Result<T, E>>,
4306 F: FnMut(T) -> U,
4307 U: IntoIterator<Item = Result<O, E>>,
4308> {
4309 inner: I,
4310 f: F,
4311 current: Option<U::IntoIter>,
4312}
4313
4314impl<
4315 T,
4316 E,
4317 O,
4318 I: Iterator<Item = Result<T, E>>,
4319 F: FnMut(T) -> U,
4320 U: IntoIterator<Item = Result<O, E>>,
4321> Iterator for FlatMapOk<T, E, O, I, F, U>
4322{
4323 type Item = Result<O, E>;
4324
4325 #[inline]
4326 fn next(&mut self) -> Option<Self::Item> {
4327 loop {
4328 if let Some(current) = &mut self.current {
4329 if let Some(next) = current.next() {
4330 return Some(next);
4331 }
4332 }
4333 self.current = None;
4334 match self.inner.next()? {
4335 Ok(e) => self.current = Some((self.f)(e).into_iter()),
4336 Err(error) => return Some(Err(error)),
4337 }
4338 }
4339 }
4340}
4341
4342enum ComparatorFunction<'a, T> {
4343 Asc(Rc<dyn Fn(&InternalTuple<T>) -> Option<ExpressionTerm> + 'a>),
4344 Desc(Rc<dyn Fn(&InternalTuple<T>) -> Option<ExpressionTerm> + 'a>),
4345}
4346
4347struct InternalTupleSet<T> {
4348 key: Vec<usize>,
4349 map: FxHashMap<u64, Vec<InternalTuple<T>>>,
4350 len: usize,
4351}
4352
4353impl<T> InternalTupleSet<T> {
4354 fn new(key: Vec<usize>) -> Self {
4355 Self {
4356 key,
4357 map: FxHashMap::default(),
4358 len: 0,
4359 }
4360 }
4361
4362 fn len(&self) -> usize {
4363 self.len
4364 }
4365
4366 fn is_empty(&self) -> bool {
4367 self.len == 0
4368 }
4369}
4370
4371impl<T: Hash> InternalTupleSet<T> {
4372 fn insert(&mut self, tuple: InternalTuple<T>) {
4373 self.map
4374 .entry(self.tuple_key(&tuple))
4375 .or_default()
4376 .push(tuple);
4377 self.len += 1;
4378 }
4379
4380 fn get(&self, tuple: &InternalTuple<T>) -> &[InternalTuple<T>] {
4381 self.map.get(&self.tuple_key(tuple)).map_or(&[], |v| v)
4382 }
4383
4384 fn tuple_key(&self, tuple: &InternalTuple<T>) -> u64 {
4385 let mut hasher = FxHasher::default();
4386 for v in &self.key {
4387 if let Some(val) = tuple.get(*v) {
4388 val.hash(&mut hasher);
4389 }
4390 }
4391 hasher.finish()
4392 }
4393}
4394
4395impl<T: Hash> Extend<InternalTuple<T>> for InternalTupleSet<T> {
4396 fn extend<I: IntoIterator<Item = InternalTuple<T>>>(&mut self, iter: I) {
4397 let iter = iter.into_iter();
4398 self.map.reserve(iter.size_hint().0);
4399 for tuple in iter {
4400 self.insert(tuple);
4401 }
4402 }
4403}
4404
4405struct StatsIterator<'a, T> {
4406 inner: InternalTuplesIterator<'a, T>,
4407 stats: Rc<EvalNodeWithStats>,
4408}
4409
4410impl<T> Iterator for StatsIterator<'_, T> {
4411 type Item = Result<InternalTuple<T>, QueryEvaluationError>;
4412
4413 fn next(&mut self) -> Option<Self::Item> {
4414 let start = Timer::now();
4415 let result = self.inner.next();
4416 let duration = start.elapsed()?;
4417 self.stats.exec_duration.set(
4418 self.stats
4419 .exec_duration
4420 .get()
4421 .and_then(|d| d.checked_add(duration)),
4422 );
4423 if matches!(result, Some(Ok(_))) {
4424 self.stats.exec_count.set(self.stats.exec_count.get() + 1);
4425 }
4426 result
4427 }
4428}
4429
4430pub struct EvalNodeWithStats {
4431 pub label: String,
4432 pub children: Vec<Rc<EvalNodeWithStats>>,
4433 pub exec_count: Cell<usize>,
4434 pub exec_duration: Cell<Option<DayTimeDuration>>,
4435}
4436
4437impl EvalNodeWithStats {
4438 pub(crate) fn empty() -> Self {
4439 Self {
4440 label: String::new(),
4441 children: Vec::new(),
4442 exec_count: Cell::new(0),
4443 exec_duration: Cell::new(None),
4444 }
4445 }
4446
4447 pub fn json_node(
4448 &self,
4449 serializer: &mut WriterJsonSerializer<impl io::Write>,
4450 with_stats: bool,
4451 ) -> io::Result<()> {
4452 serializer.serialize_event(JsonEvent::StartObject)?;
4453 serializer.serialize_event(JsonEvent::ObjectKey("name".into()))?;
4454 serializer.serialize_event(JsonEvent::String((&self.label).into()))?;
4455 if with_stats {
4456 serializer.serialize_event(JsonEvent::ObjectKey("number of results".into()))?;
4457 serializer
4458 .serialize_event(JsonEvent::Number(self.exec_count.get().to_string().into()))?;
4459 if let Some(duration) = self.exec_duration.get() {
4460 serializer.serialize_event(JsonEvent::ObjectKey("duration in seconds".into()))?;
4461 serializer
4462 .serialize_event(JsonEvent::Number(duration.as_seconds().to_string().into()))?;
4463 }
4464 }
4465 serializer.serialize_event(JsonEvent::ObjectKey("children".into()))?;
4466 serializer.serialize_event(JsonEvent::StartArray)?;
4467 for child in &self.children {
4468 child.json_node(serializer, with_stats)?;
4469 }
4470 serializer.serialize_event(JsonEvent::EndArray)?;
4471 serializer.serialize_event(JsonEvent::EndObject)
4472 }
4473}
4474
4475impl fmt::Debug for EvalNodeWithStats {
4476 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
4477 let mut obj = f.debug_struct("Node");
4478 obj.field("name", &self.label);
4479 if let Some(exec_duration) = self.exec_duration.get() {
4480 obj.field("number of results", &self.exec_count.get());
4481 obj.field(
4482 "duration in seconds",
4483 &f32::from(Float::from(exec_duration.as_seconds())),
4484 );
4485 }
4486 if !self.children.is_empty() {
4487 obj.field("children", &self.children);
4488 }
4489 obj.finish()
4490 }
4491}
4492
4493fn eval_node_label(node: &GraphPattern) -> String {
4494 match node {
4495 GraphPattern::Distinct { .. } => "Distinct(Hash)".to_owned(),
4496 GraphPattern::Extend {
4497 expression,
4498 variable,
4499 ..
4500 } => format!(
4501 "Extend({} -> {variable})",
4502 spargebra::algebra::Expression::from(expression)
4503 ),
4504 GraphPattern::Filter { expression, .. } => format!(
4505 "Filter({})",
4506 spargebra::algebra::Expression::from(expression)
4507 ),
4508 GraphPattern::Graph { graph_name } => format!("Graph({graph_name})"),
4509 GraphPattern::Group {
4510 variables,
4511 aggregates,
4512 ..
4513 } => {
4514 format!(
4515 "Aggregate({})",
4516 format_list(variables.iter().map(ToString::to_string).chain(
4517 aggregates.iter().map(|(v, agg)| format!(
4518 "{} -> {v}",
4519 spargebra::algebra::AggregateExpression::from(agg)
4520 ))
4521 ))
4522 )
4523 }
4524 GraphPattern::Join { algorithm, .. } => match algorithm {
4525 JoinAlgorithm::HashBuildLeftProbeRight { keys } => format!(
4526 "LeftJoin(HashBuildLeftProbeRight, keys = {})",
4527 format_list(keys)
4528 ),
4529 },
4530 #[cfg(feature = "sep-0006")]
4531 GraphPattern::Lateral { right, .. } => {
4532 if let GraphPattern::LeftJoin {
4533 left: nested_left,
4534 expression,
4535 ..
4536 } = right.as_ref()
4537 {
4538 if nested_left.is_empty_singleton() {
4539 return format!(
4541 "ForLoopLeftJoin(expression = {})",
4542 spargebra::algebra::Expression::from(expression)
4543 );
4544 }
4545 }
4546 "Lateral".to_owned()
4547 }
4548 GraphPattern::LeftJoin {
4549 algorithm,
4550 expression,
4551 ..
4552 } => match algorithm {
4553 LeftJoinAlgorithm::HashBuildRightProbeLeft { keys } => format!(
4554 "LeftJoin(HashBuildRightProbeLeft, keys = {}, expression = {})",
4555 format_list(keys),
4556 spargebra::algebra::Expression::from(expression)
4557 ),
4558 },
4559 GraphPattern::Minus { algorithm, .. } => match algorithm {
4560 MinusAlgorithm::HashBuildRightProbeLeft { keys } => format!(
4561 "AntiJoin(HashBuildRightProbeLeft, keys = {})",
4562 format_list(keys)
4563 ),
4564 },
4565 GraphPattern::OrderBy { expression, .. } => {
4566 format!(
4567 "Sort({})",
4568 format_list(
4569 expression
4570 .iter()
4571 .map(spargebra::algebra::OrderExpression::from)
4572 )
4573 )
4574 }
4575 GraphPattern::Path {
4576 subject,
4577 path,
4578 object,
4579 graph_name,
4580 } => {
4581 if let Some(graph_name) = graph_name {
4582 format!("Path({subject} {path} {object} {graph_name})")
4583 } else {
4584 format!("Path({subject} {path} {object})")
4585 }
4586 }
4587 GraphPattern::Project { variables, .. } => {
4588 format!("Project({})", format_list(variables))
4589 }
4590 GraphPattern::QuadPattern {
4591 subject,
4592 predicate,
4593 object,
4594 graph_name,
4595 } => {
4596 if let Some(graph_name) = graph_name {
4597 format!("QuadPattern({subject} {predicate} {object} {graph_name})")
4598 } else {
4599 format!("QuadPattern({subject} {predicate} {object})")
4600 }
4601 }
4602 GraphPattern::Reduced { .. } => "Reduced".to_owned(),
4603 GraphPattern::Service { name, silent, .. } => {
4604 if *silent {
4605 format!("Service({name}, Silent)")
4606 } else {
4607 format!("Service({name})")
4608 }
4609 }
4610 GraphPattern::Slice { start, length, .. } => {
4611 if let Some(length) = length {
4612 format!("Slice(start = {start}, length = {length})")
4613 } else {
4614 format!("Slice(start = {start})")
4615 }
4616 }
4617 GraphPattern::Union { .. } => "Union".to_owned(),
4618 GraphPattern::Values {
4619 variables,
4620 bindings,
4621 } => {
4622 format!(
4623 "StaticBindings(({}), ({}))",
4624 format_list(variables),
4625 format_list(bindings.iter().map(|b| {
4626 format!(
4627 "({})",
4628 format_list(b.iter().map(|t| {
4629 t.as_ref()
4630 .map_or_else(|| "UNDEF".into(), GroundTerm::to_string)
4631 }))
4632 )
4633 }))
4634 )
4635 }
4636 }
4637}
4638
4639fn format_list<T: ToString>(values: impl IntoIterator<Item = T>) -> String {
4640 values
4641 .into_iter()
4642 .map(|v| v.to_string())
4643 .collect::<Vec<_>>()
4644 .join(", ")
4645}
4646
4647pub struct Timer {
4648 start: DateTime,
4649}
4650
4651impl Timer {
4652 pub fn now() -> Self {
4653 Self {
4654 start: DateTime::now(),
4655 }
4656 }
4657
4658 pub fn elapsed(&self) -> Option<DayTimeDuration> {
4659 DateTime::now().checked_sub(self.start)
4660 }
4661}
4662
4663#[derive(Clone, Default)]
4667pub struct CancellationToken {
4668 value: Arc<AtomicBool>,
4669}
4670
4671impl CancellationToken {
4672 #[inline]
4673 pub fn new() -> Self {
4674 Self {
4675 value: Arc::new(AtomicBool::new(false)),
4676 }
4677 }
4678
4679 #[inline]
4680 pub fn cancel(&self) {
4681 self.value.store(true, atomic::Ordering::Relaxed);
4682 }
4683
4684 #[inline]
4685 pub fn is_cancelled(&self) -> bool {
4686 self.value.load(atomic::Ordering::Relaxed)
4687 }
4688
4689 fn ensure_alive(&self) -> Result<(), QueryEvaluationError> {
4690 if self.is_cancelled() {
4691 Err(QueryEvaluationError::Cancelled)
4692 } else {
4693 Ok(())
4694 }
4695 }
4696}