Coverage for src/time_agnostic_library/agnostic_query.py: 99%
853 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-10-10 10:04 +0000
1# SPDX-FileCopyrightText: 2021-2026 Arcangelo Massari <arcangelo.massari@unibo.it>
2#
3# SPDX-License-Identifier: ISC
6import atexit
7import json
8import os
9import re
10from collections import Counter
11from collections.abc import Callable, Iterable
12from concurrent.futures import ThreadPoolExecutor, as_completed
13from itertools import pairwise, product
14from pathlib import Path
15from typing import NoReturn, cast
17from rdflib import Literal, URIRef
18from rdflib.paths import InvPath
19from rdflib.paths import Path as PropertyPath
20from rdflib.plugins.sparql.parserutils import CompValue
21from rdflib.plugins.sparql.processor import prepareQuery
23from time_agnostic_library.agnostic_entity import (
24 AgnosticEntity,
25 _fast_parse_update,
26 _iter_working_states,
27 _materialize_versions,
28 _parse_datetime,
29 _select_interval_snapshots,
30)
31from time_agnostic_library.prov_entity import ProvEntity
32from time_agnostic_library.qlever import QLEVER_HAS_WORD, qlever_search_tokens
33from time_agnostic_library.sparql import Sparql, _binding_to_n3, _n3_to_binding
34from time_agnostic_library.support import convert_to_datetime
36CONFIG_PATH = "./config.json"
38_OBJECT_POS = 2
40_PARALLEL_THRESHOLD = os.cpu_count() or 1
42# jena-text answers a text:query with no explicit limit with at most 10000 hits
43# and drops the rest silently, leaving entities out of the reconstruction.
44_FUSEKI_TEXT_SEARCH_LIMIT = 10_000_000
46# Algebra nodes the pattern collection below knows how to walk. Anything else
47# would be flattened into a conjunction of its operands, silently returning
48# wrong results, so it is rejected instead.
49_SUPPORTED_ALGEBRA_NODES = frozenset(
50 {"SelectQuery", "Project", "Distinct", "Join", "LeftJoin", "BGP"}
51)
53# rdflib maps BIND, the expressions of a SELECT clause and every form of
54# grouping onto Extend, which always wraps the aggregation nodes below it, so
55# the label has to hold for all of them.
56_SPARQL_CONSTRUCT_NAMES = {
57 "Union": "UNION",
58 "Filter": "FILTER",
59 "Minus": "MINUS",
60 "Graph": "GRAPH",
61 "OrderBy": "ORDER BY",
62 "Slice": "LIMIT or OFFSET",
63 "Extend": "BIND or an aggregate function",
64 "ToMultiSet": "VALUES or a subquery",
65}
67_COVERAGE_NOTE = (
68 "Time agnostic queries cover SELECT queries made of basic graph patterns "
69 "and OPTIONAL clauses."
70)
73def _reject(construct: str) -> NoReturn:
74 msg = f"The query uses {construct}, which is not supported. {_COVERAGE_NOTE}"
75 raise ValueError(msg)
78def _reject_unsupported(node_name: str) -> None:
79 if node_name in _SUPPORTED_ALGEBRA_NODES:
80 return
81 _reject(_SPARQL_CONSTRUCT_NAMES.get(node_name, node_name))
84def _contains_algebra_node(node: CompValue, name: str) -> bool:
85 if node.name == name:
86 return True
87 return any(
88 _contains_algebra_node(value, name)
89 for value in node.values()
90 if isinstance(value, CompValue)
91 )
94def _is_unsupported_path(term: object) -> bool:
95 # An inverse path over a plain predicate is the one path form the library
96 # resolves, by rewriting it in _n3_triples. Every other form would reach the
97 # matcher as an opaque predicate string and match nothing.
98 if isinstance(term, InvPath):
99 return not isinstance(term.arg, URIRef)
100 return isinstance(term, PropertyPath)
103def _n3_triples(node: CompValue) -> list[tuple[str, ...]]:
104 triples = []
105 for triple in node["triples"]:
106 if any(_is_unsupported_path(el) for el in triple):
107 _reject("a property path")
108 subject, predicate, obj = triple
109 if isinstance(predicate, InvPath):
110 # Read the pattern in its direct orientation, so that every step
111 # downstream works on a plain triple.
112 subject, predicate, obj = obj, predicate.arg, subject
113 triples.append((subject.n3(), predicate.n3(), obj.n3()))
114 return triples
117def _reject_filter_in_optional(node: CompValue) -> None:
118 # rdflib keeps the condition of a FILTER inside an OPTIONAL in the expr slot
119 # of the left join, where TrueFilter stands for no condition at all.
120 if node["expr"].name != "TrueFilter":
121 _reject("a FILTER inside an OPTIONAL")
124_IO_EXECUTOR = ThreadPoolExecutor(max_workers=2)
125atexit.register(_IO_EXECUTOR.shutdown, wait=False)
128def _run_in_parallel(worker_fn, args_list):
129 if len(args_list) < _PARALLEL_THRESHOLD:
130 for args in args_list:
131 yield worker_fn(*args)
132 return
133 with ThreadPoolExecutor() as executor:
134 futures = {
135 executor.submit(worker_fn, *args): i for i, args in enumerate(args_list)
136 }
137 for future in as_completed(futures):
138 yield future.result()
141def _reconstruct_entity_worker(entity, config, on_time):
142 agnostic_entity = AgnosticEntity(
143 entity,
144 config=config,
145 include_related_objects=False,
146 include_merged_entities=False,
147 include_reverse_relations=False,
148 )
149 if on_time:
150 entity_graphs, _, _ = agnostic_entity.get_state_at_time(
151 time=on_time,
152 include_prov_metadata=False,
153 )
154 return entity, entity_graphs
155 entity_history = agnostic_entity.get_history(include_prov_metadata=False)
156 return entity, entity_history[0]
159_LITERAL_N3_RE = re.compile(
160 r"""
161 ^ " (?P<lexical>.*) " # Greedy matching retains escaped quotes.
162 (?: @(?P<language>[\w-]+) | \^\^<[^>]*> )?
163 \Z
164 """,
165 re.VERBOSE,
166)
169def _normalize_constant(term_n3: str) -> str | None:
170 if term_n3.startswith("<") and term_n3.endswith(">"):
171 return term_n3
172 match = _LITERAL_N3_RE.match(term_n3)
173 if match is None:
174 return None
175 language = match["language"]
176 if language is None:
177 return term_n3
178 return f'"{match["lexical"]}"@{language.lower()}'
181def _pattern_constants(triple: tuple) -> set[str]:
182 constants = (_normalize_constant(el) for el in triple[:3])
183 return {constant for constant in constants if constant is not None}
186def _pattern_search_terms(triple: tuple) -> set[str]:
187 return {el[1:-1] for el in triple[:3] if el.startswith("<") and el.endswith(">")}
190def _fuseki_search_phrases(triple: tuple) -> list[str]:
191 predicate = _normalize_constant(triple[1])
192 obj = _normalize_constant(triple[2])
193 if obj is not None and obj.startswith("<"):
194 if predicate is not None:
195 return [f"{predicate} {obj} ."]
196 return [f"{obj} ."]
197 return [f"<{term}>" for term in sorted(_pattern_search_terms(triple))]
200def _escape_search_term(text: str, *quotes: str) -> str:
201 escaped = text.replace("\\", "\\\\")
202 for quote in quotes:
203 escaped = escaped.replace(quote, "\\" + quote)
204 return escaped
207def _expected_quad_slots(triple: tuple) -> tuple[str | None, str | None, str | None]:
208 subject, predicate, obj = triple[:3]
209 return (
210 _normalize_constant(subject),
211 _normalize_constant(predicate),
212 _normalize_constant(obj),
213 )
216def _matching_update_quads(
217 update_query: str, triple: tuple
218) -> list[tuple[str, str, str, str]]:
219 quad_matches = _quad_filter_for_pattern(triple)
220 return [
221 quad
222 for _, quads in _fast_parse_update(update_query)
223 for quad in quads
224 if quad_matches(quad)
225 ]
228def _quad_filter_for_pattern(
229 triple: tuple,
230) -> Callable[[tuple[str, ...]], bool]:
231 expected = tuple(
232 (index, slot, slot.startswith("<"))
233 for index, slot in enumerate(_expected_quad_slots(triple))
234 if slot is not None
235 )
237 def matches(quad: tuple[str, ...]) -> bool:
238 return all(
239 quad[index] == slot if is_iri else _normalize_constant(quad[index]) == slot
240 for index, slot, is_iri in expected
241 )
243 return matches
246def _sparql_values(uris: set[str]) -> str:
247 return " ".join(f"<{uri}>" for uri in uris)
250def _wrap_in_graph(body: str, *, is_quadstore: bool) -> str:
251 if is_quadstore:
252 return f"GRAPH ?g {{ {body} }}"
253 return body
256def _batch_query_provenance_snapshots(
257 entity_uris: set[str], config: dict, start_time: str | None = None
258) -> dict[str, list[dict]]:
259 # Two joins instead of one OPTIONAL, no GRAPH around them, and the time
260 # bound applied here: QLever evaluates an OPTIONAL or a FILTER over the
261 # whole provenance before it joins the VALUES, and two patterns under the
262 # same GRAPH variable join on the graph too, which sorts every snapshot.
263 # The union default graph answers the query, as for the other provenance
264 # queries.
265 snapshots = f"""
266 ?snapshot <{ProvEntity.iri_specialization_of}> ?entity;
267 <{ProvEntity.iri_generated_at_time}> ?time.
268 VALUES ?entity {{ {_sparql_values(entity_uris)} }}
269 """
270 updates = f"""
271 ?snapshot <{ProvEntity.iri_specialization_of}> ?entity;
272 <{ProvEntity.iri_has_update_query}> ?updateQuery.
273 VALUES ?entity {{ {_sparql_values(entity_uris)} }}
274 """
275 snapshot_rows = Sparql(
276 f"SELECT ?snapshot ?entity ?time WHERE {{ {snapshots} }}", config
277 ).run_select_query()
278 update_rows = Sparql(
279 f"SELECT ?snapshot ?updateQuery WHERE {{ {updates} }}", config
280 ).run_select_query()
281 update_queries = {
282 binding["snapshot"]["value"]: binding["updateQuery"]["value"]
283 for binding in update_rows["results"]["bindings"]
284 }
285 start = _parse_datetime(start_time) if start_time is not None else None
286 output: dict[str, list[dict]] = {uri: [] for uri in entity_uris}
287 for binding in snapshot_rows["results"]["bindings"]:
288 snapshot = binding["snapshot"]["value"]
289 time = binding["time"]["value"]
290 after_start = start is None or _parse_datetime(time) > start
291 entry = {
292 "time": time,
293 "updateQuery": update_queries[snapshot]
294 if snapshot in update_queries and after_start
295 else None,
296 }
297 output[binding["entity"]["value"]].append(entry)
298 return output
301def _sparql_filter_in(var: str, uris: set[str]) -> str:
302 return f"FILTER({var} IN ({', '.join(f'<{uri}>' for uri in uris)}))"
305def _batch_query_dataset_triples(
306 entity_uris: set[str],
307 config: dict,
308 *,
309 is_virtuoso: bool,
310 triple: tuple | None = None,
311) -> dict[str, set[tuple]]:
312 is_quadstore = config["dataset"]["is_quadstore"]
313 predicate = "?p"
314 obj = "?o"
315 if triple is not None:
316 predicate = triple[1] if _normalize_constant(triple[1]) is not None else "?p"
317 obj = triple[2] if _normalize_constant(triple[2]) is not None else "?o"
318 # Virtuoso resolves VALUES inexplicably slowly, while Jena needs it: under
319 # FILTER ... IN no index prefix applies and it scans the whole store.
320 if is_virtuoso:
321 body = f"?s {predicate} {obj}. {_sparql_filter_in('?s', entity_uris)}"
322 else:
323 body = f"VALUES ?s {{ {_sparql_values(entity_uris)} }} ?s {predicate} {obj}."
324 wrapped = _wrap_in_graph(body, is_quadstore=is_quadstore)
325 select_vars = " ".join(
326 variable
327 for variable in ("?s", predicate, obj, "?g" if is_quadstore else None)
328 if variable is not None and variable.startswith("?")
329 )
330 query = f"SELECT {select_vars} WHERE {{ {wrapped} }}"
331 results = Sparql(query, config).run_select_query()
332 output: dict[str, set[tuple]] = {uri: set() for uri in entity_uris}
333 for binding in results["results"]["bindings"]:
334 s_val = binding["s"]["value"]
335 s = _binding_to_n3(binding["s"])
336 p = _binding_to_n3(binding["p"]) if predicate == "?p" else predicate
337 o = _binding_to_n3(binding["o"]) if obj == "?o" else obj
338 if is_quadstore and "g" in binding:
339 output[s_val].add((s, p, o, _binding_to_n3(binding["g"])))
340 else:
341 output[s_val].add((s, p, o))
342 return output
345def _reconstruct_at_time_as_sets(
346 prov_snapshots: list[dict],
347 dataset_quads: set[tuple],
348 on_time: tuple[str | None, str | None],
349 triple: tuple | None = None,
350) -> list[tuple[str, tuple]]:
351 if not prov_snapshots:
352 return []
353 sorted_snaps = sorted(
354 prov_snapshots, key=lambda x: _parse_datetime(x["time"]), reverse=True
355 )
356 snapshot_bindings = [
357 {"time": {"value": snapshot["time"]}} for snapshot in sorted_snaps
358 ]
359 relevant, start_timestamp_alias = _select_interval_snapshots(
360 on_time, snapshot_bindings, time_index="time"
361 )
362 if not relevant:
363 return []
364 relevant_times = {r["time"]["value"] for r in relevant}
365 sorted_versions = [
366 (snapshot["time"], snapshot["updateQuery"]) for snapshot in sorted_snaps
367 ]
368 quad_filter = _quad_filter_for_pattern(triple) if triple is not None else None
369 materialized = _materialize_versions(
370 sorted_versions, dataset_quads, relevant_times, quad_filter
371 )
372 return [
373 (
374 start_timestamp_alias[1]
375 if start_timestamp_alias is not None
376 and timestamp == start_timestamp_alias[0]
377 else timestamp,
378 quads,
379 )
380 for timestamp, quads in materialized
381 ]
384def _match_single_pattern(
385 triple_pattern: tuple, quads: Iterable[tuple[str, ...]]
386) -> list[dict]:
387 s_pat, p_pat, o_pat = triple_pattern[0], triple_pattern[1], triple_pattern[2]
388 s_is_var = s_pat.startswith("?")
389 p_is_var = p_pat.startswith("?")
390 o_is_var = o_pat.startswith("?")
391 bindings = []
392 for quad in quads:
393 s, p, o = quad[0], quad[1], quad[2]
394 if not p_is_var and p != p_pat:
395 continue
396 if not o_is_var and o != o_pat:
397 continue
398 if not s_is_var and s != s_pat:
399 continue
400 binding = {}
401 if s_is_var:
402 binding[s_pat[1:]] = _n3_to_binding(s)
403 if p_is_var:
404 binding[p_pat[1:]] = _n3_to_binding(p)
405 if o_is_var:
406 binding[o_pat[1:]] = _n3_to_binding(o)
407 bindings.append(binding)
408 return bindings
411def _index_quads_by_subject(
412 quads: set[tuple[str, ...]],
413) -> dict[str, set[tuple[str, ...]]]:
414 by_subject: dict[str, set[tuple[str, ...]]] = {}
415 for quad in quads:
416 subject = quad[0]
417 if subject in by_subject:
418 by_subject[subject].add(quad)
419 else:
420 by_subject[subject] = {quad}
421 return by_subject
424def _merge_entity_bindings(
425 entity_bindings: dict[str, dict[str, list[dict]]],
426) -> dict[str, list[dict]]:
427 all_timestamps: set[str] = set()
428 for per_ts in entity_bindings.values():
429 all_timestamps.update(per_ts.keys())
430 sorted_timestamps = sorted(all_timestamps, key=_parse_datetime)
431 result: dict[str, list[dict]] = {}
432 last_known: dict[str, list[dict]] = {}
433 for ts in sorted_timestamps:
434 merged: list[dict] = []
435 for entity_str, per_ts in entity_bindings.items():
436 if ts in per_ts:
437 last_known[entity_str] = per_ts[ts]
438 if entity_str in last_known:
439 merged.extend(last_known[entity_str])
440 result[ts] = merged
441 return result
444def _binding_key(
445 binding: dict[str, dict[str, str]],
446) -> frozenset[tuple[str, frozenset[tuple[str, str]]]]:
447 return frozenset(
448 (variable, frozenset(value.items())) for variable, value in binding.items()
449 )
452def _bag_difference(left: list[dict], right: list[dict]) -> list[dict]:
453 remaining = Counter(_binding_key(binding) for binding in right)
454 difference = []
455 for binding in left:
456 key = _binding_key(binding)
457 if remaining[key] > 0:
458 remaining[key] -= 1
459 else:
460 difference.append(binding)
461 return difference
464def _states_at(
465 history: list[tuple[str, list[dict]]], timeline: list[str]
466) -> list[tuple[str, list[dict]]]:
467 history_position = 0
468 state: list[dict] = []
469 states = []
470 for timestamp in timeline:
471 requested = _parse_datetime(timestamp)
472 while (
473 history_position < len(history)
474 and _parse_datetime(history[history_position][0]) <= requested
475 ):
476 state = history[history_position][1]
477 history_position += 1
478 states.append((timestamp, state))
479 return states
482def _build_solution_delta(
483 results: dict[str, list[dict]],
484 on_time: tuple[str | None, str | None] | None,
485) -> dict[str, list | None]:
486 history = sorted(results.items(), key=lambda item: _parse_datetime(item[0]))
487 if not history:
488 return {"additions": [], "deletions": [], "changes": []}
489 start = on_time[0] if on_time and on_time[0] else history[0][0]
490 end = on_time[1] if on_time and on_time[1] else history[-1][0]
491 if _parse_datetime(start) > _parse_datetime(end):
492 message = "The start of the interval must not follow its end"
493 raise ValueError(message)
494 timeline = [start]
495 timeline.extend(
496 timestamp
497 for timestamp, _ in history
498 if _parse_datetime(start) < _parse_datetime(timestamp) < _parse_datetime(end)
499 )
500 if end != start:
501 timeline.append(end)
502 states = _states_at(history, timeline)
503 start_state = states[0][1]
504 end_state = states[-1][1]
505 changes = [
506 {
507 "start": previous_time,
508 "end": current_time,
509 "additions": _bag_difference(current_state, previous_state),
510 "deletions": _bag_difference(previous_state, current_state),
511 }
512 for (previous_time, previous_state), (current_time, current_state) in pairwise(
513 states
514 )
515 ]
516 return {
517 "additions": _bag_difference(end_state, start_state),
518 "deletions": _bag_difference(start_state, end_state),
519 "changes": changes,
520 }
523class AgnosticQuery:
524 blazegraph_full_text_search: bool
525 fuseki_full_text_search: bool
526 virtuoso_full_text_search: bool
527 qlever_full_text_search: bool
528 graphdb_connector_name: str
530 def __init__(
531 self,
532 query: str,
533 on_time: tuple[str | None, str | None] | None = (None, None),
534 *,
535 merge_aware: bool = False,
536 include_prov_metadata: bool = False,
537 config_path: str = CONFIG_PATH,
538 config_dict: dict | None = None,
539 ):
540 self.query = query
541 self.merge_aware = merge_aware
542 self.include_prov_metadata = include_prov_metadata
543 self.config_path = config_path
544 self._merge_adjacency: dict[str, set[str]] = {}
545 self._merge_events: dict[str, dict] = {}
546 self._merge_scanned_entities: set[str] = set()
547 if config_dict is not None:
548 self.config = config_dict
549 else:
550 with Path(config_path).open(encoding="utf8") as json_file:
551 self.config = json.load(json_file)
552 self.__init_text_index(self.config)
553 if on_time:
554 after_time = convert_to_datetime(on_time[0], stringify=True)
555 before_time = convert_to_datetime(on_time[1], stringify=True)
556 self.on_time: tuple[str | None, str | None] | None = (
557 after_time,
558 before_time,
559 ) # type: ignore[assignment]
560 else:
561 self.on_time = None
562 self.reconstructed_entities: set[str] = set()
563 self.vars_to_explicit_by_time: dict = {}
564 self.relevant_entities_graphs: dict[str, dict[str, set]] = {}
565 self.relevant_graphs: dict[str, set[tuple[str, ...]]] = {}
566 self._rebuild_relevant_graphs()
568 def _query_adjacent_merge_events(self, entity_uris: set[str]) -> list[dict]:
569 values = _sparql_values(entity_uris)
570 merge_pattern = f"""
571 ?snapshot <{ProvEntity.iri_specialization_of}> ?survivor;
572 <{ProvEntity.iri_generated_at_time}> ?time;
573 <{ProvEntity.iri_was_derived_from}> ?sourceSnapshot.
574 """
575 source_pattern = (
576 f"?sourceSnapshot <{ProvEntity.iri_specialization_of}> ?absorbed."
577 )
578 if self.config["provenance"]["is_quadstore"]:
579 merge_pattern = f"GRAPH ?mergeGraph {{ {merge_pattern} }}"
580 source_pattern = f"GRAPH ?sourceGraph {{ {source_pattern} }}"
581 query = f"""
582 SELECT DISTINCT ?snapshot ?time ?survivor ?absorbed
583 WHERE {{
584 VALUES ?participant {{ {values} }}
585 {merge_pattern}
586 {source_pattern}
587 FILTER (?survivor != ?absorbed)
588 FILTER (?survivor = ?participant || ?absorbed = ?participant)
589 }}
590 """
591 results = Sparql(query, self.config).run_select_query()
592 return results["results"]["bindings"]
594 def _expand_entities_with_merges(self, entity_uris: set[str]) -> set[str]:
595 if not self.merge_aware or not entity_uris:
596 return set(entity_uris)
597 component = set(entity_uris)
598 frontier = component.difference(self._merge_scanned_entities)
599 while frontier:
600 self._merge_scanned_entities.update(frontier)
601 discovered = set()
602 for binding in self._query_adjacent_merge_events(frontier):
603 survivor = binding["survivor"]["value"]
604 absorbed = binding["absorbed"]["value"]
605 snapshot = binding["snapshot"]["value"]
606 event = self._merge_events.setdefault(
607 snapshot,
608 {
609 "time": str(
610 convert_to_datetime(
611 binding["time"]["value"], stringify=True
612 )
613 ),
614 "snapshot": snapshot,
615 "survivor": survivor,
616 "absorbed": set(),
617 },
618 )
619 event["absorbed"].add(absorbed)
620 self._merge_adjacency.setdefault(survivor, set()).add(absorbed)
621 self._merge_adjacency.setdefault(absorbed, set()).add(survivor)
622 discovered.update((survivor, absorbed))
623 component.update(discovered)
624 frontier = discovered.difference(self._merge_scanned_entities)
625 queue = list(component)
626 while queue:
627 entity_uri = queue.pop()
628 adjacent_entities = (
629 self._merge_adjacency[entity_uri]
630 if entity_uri in self._merge_adjacency
631 else set()
632 )
633 for adjacent in adjacent_entities:
634 if adjacent not in component:
635 component.add(adjacent)
636 queue.append(adjacent)
637 return component
639 def _entity_aliases(self, entity_uri: str) -> set[str]:
640 return self._expand_entities_with_merges({entity_uri})
642 def _expand_triple_aliases(self, triple: tuple[str, ...]) -> list[tuple[str, ...]]:
643 term_options = []
644 for index, term in enumerate(triple):
645 if (
646 index in (0, _OBJECT_POS)
647 and term.startswith("<")
648 and term.endswith(">")
649 ):
650 term_options.append(
651 [f"<{uri}>" for uri in sorted(self._entity_aliases(term[1:-1]))]
652 )
653 else:
654 term_options.append([term])
655 return [tuple(terms) for terms in product(*term_options)]
657 def _load_provenance(
658 self, entity_uris: set[str]
659 ) -> tuple[dict | None, dict | None]:
660 if not self.include_prov_metadata:
661 return None, None
662 provenance = {}
663 other_provenance = {}
664 for entity_uri in sorted(entity_uris):
665 agnostic_entity = AgnosticEntity(
666 entity_uri,
667 config=self.config,
668 include_related_objects=False,
669 include_merged_entities=False,
670 include_reverse_relations=False,
671 )
672 if self.on_time is None:
673 _, entity_metadata = agnostic_entity.get_history(
674 include_prov_metadata=True
675 )
676 snapshots = (
677 entity_metadata[entity_uri] if entity_uri in entity_metadata else {}
678 )
679 if snapshots:
680 provenance[entity_uri] = snapshots
681 continue
682 _, relevant_snapshots, other_snapshots = agnostic_entity.get_state_at_time(
683 self.on_time,
684 include_prov_metadata=True,
685 )
686 if relevant_snapshots:
687 provenance[entity_uri] = relevant_snapshots
688 if other_snapshots:
689 other_provenance[entity_uri] = other_snapshots
690 return provenance, other_provenance
692 def __init_text_index(self, config: dict):
693 config = {
694 "blazegraph_full_text_search": "no",
695 "fuseki_full_text_search": "no",
696 "virtuoso_full_text_search": "no",
697 "qlever_full_text_search": "no",
698 "graphdb_connector_name": "",
699 **config,
700 }
701 for full_text_search in (
702 "blazegraph_full_text_search",
703 "fuseki_full_text_search",
704 "virtuoso_full_text_search",
705 "qlever_full_text_search",
706 ):
707 ts_full_text_search: str = config[full_text_search]
708 if ts_full_text_search.lower() in {"true", "1", 1, "t", "y", "yes", "ok"}:
709 setattr(self, full_text_search, True)
710 elif (
711 ts_full_text_search.lower() in {"false", "0", 0, "n", "f", "no"}
712 or not ts_full_text_search
713 ):
714 setattr(self, full_text_search, False)
715 else:
716 msg = (
717 f"Enter a valid value for '{full_text_search}' in the "
718 "configuration file, for example 'yes' or 'no'."
719 )
720 raise ValueError(msg)
721 self.graphdb_connector_name = config["graphdb_connector_name"]
722 if (
723 len(
724 [
725 index
726 for index in [
727 self.blazegraph_full_text_search,
728 self.fuseki_full_text_search,
729 self.virtuoso_full_text_search,
730 self.qlever_full_text_search,
731 self.graphdb_connector_name,
732 ]
733 if index
734 ]
735 )
736 > 1
737 ):
738 msg = (
739 "The use of multiple indexing systems simultaneously "
740 "is currently not supported."
741 )
742 raise ValueError(msg)
744 def _process_query(self) -> list[tuple[str, ...]]:
745 # Parse the SPARQL string into an algebra tree via rdflib, then walk
746 # the tree to extract triple patterns as N3 strings, separating
747 # mandatory patterns from OPTIONAL groups.
748 algebra = prepareQuery(self.query).algebra
749 if algebra.name != "SelectQuery":
750 msg = "Only SELECT queries are allowed."
751 raise ValueError(msg)
752 mandatory: list[tuple[str, ...]] = []
753 self._optional_groups: list[list[tuple[str, ...]]] = []
754 self._collect_patterns(algebra, mandatory)
755 all_triples = list(mandatory)
756 for group in self._optional_groups:
757 all_triples.extend(group)
758 # Reject triples made of only variables (e.g. ?s ?p ?o): they would
759 # match every entity in the dataset, making the query too expensive.
760 triples_without_hook = [
761 t for t in all_triples if all(el.startswith("?") for el in t)
762 ]
763 if triples_without_hook:
764 msg = (
765 "Could not perform a generic time agnostic query. "
766 "Please, specify at least one URI or Literal within the query."
767 )
768 raise ValueError(msg)
769 self._select_vars = [str(v) for v in algebra["PV"]]
770 self._distinct = _contains_algebra_node(algebra, "Distinct")
771 self._mandatory_triples = mandatory
772 return all_triples
774 def _collect_patterns(
775 self, node: CompValue, mandatory: list[tuple[str, ...]]
776 ) -> None:
777 name = node.name
778 if name == "LeftJoin":
779 # OPTIONAL = left join: p1 (left, mandatory) must match, p2 (right,
780 # optional) extends the binding if possible, otherwise it's ignored
781 _reject_filter_in_optional(node)
782 self._collect_patterns(node["p1"], mandatory)
783 opt_group: list[tuple[str, ...]] = []
784 self._collect_triples_flat(node["p2"], opt_group)
785 if opt_group:
786 self._optional_groups.append(opt_group)
787 elif name == "Join":
788 # Both sides are mandatory (rdflib splits BGPs into Join nodes)
789 self._collect_patterns(node["p1"], mandatory)
790 self._collect_patterns(node["p2"], mandatory)
791 elif "triples" in node:
792 # BGP leaf node: convert rdflib terms to N3 strings
793 mandatory.extend(_n3_triples(node))
794 else:
795 _reject_unsupported(name)
796 for v in node.values():
797 if isinstance(v, CompValue):
798 self._collect_patterns(v, mandatory)
800 def _collect_triples_flat(
801 self, node: CompValue, triples: list[tuple[str, ...]]
802 ) -> None:
803 name = node.name
804 if name == "LeftJoin":
805 # A nested OPTIONAL: its patterns join the group being flattened.
806 _reject_filter_in_optional(node)
807 self._collect_triples_flat(node["p1"], triples)
808 self._collect_triples_flat(node["p2"], triples)
809 return
810 if "triples" in node:
811 triples.extend(_n3_triples(node))
812 return
813 _reject_unsupported(name)
814 for v in node.values():
815 if isinstance(v, CompValue):
816 self._collect_triples_flat(v, triples)
818 def _rebuild_relevant_graphs(self) -> None:
819 triples_checked = set()
820 all_isolated = True
821 self.triples = self._process_query()
822 for triple in self.triples:
823 if self._is_isolated(triple) and self._is_a_new_triple(
824 triple, triples_checked
825 ):
826 present_entities = self._get_present_entities(triple)
827 self._rebuild_relevant_entity(triple[0])
828 self._find_entities_in_update_queries(triple, present_entities)
829 else:
830 all_isolated = False
831 self._rebuild_relevant_entity(triple[0])
832 triples_checked.add(triple)
833 self._align_snapshots()
834 if not all_isolated:
835 self._solve_variables()
837 def _is_isolated(self, triple: tuple) -> bool:
838 if triple[0].startswith("<") and triple[0].endswith(">"):
839 return False
840 variables = [el for el in triple if el.startswith("?")]
841 for variable in variables:
842 other_triples = {t for t in self.triples if t != triple}
843 if self._there_is_transitive_closure(variable, other_triples):
844 return False
845 return True
847 def _there_is_transitive_closure(self, variable: str, triples: set[tuple]) -> bool:
848 there_is_transitive_closure = False
849 for triple in triples:
850 if variable in triple and triple.index(variable) == _OBJECT_POS:
851 if triple[0].startswith("<") and triple[0].endswith(">"):
852 return True
853 if triple[0].startswith("?"):
854 other_triples = {t for t in triples if t != triple}
855 there_is_transitive_closure = self._there_is_transitive_closure(
856 triple[0], other_triples
857 )
858 return there_is_transitive_closure
860 def _rebuild_relevant_entity(self, entity_n3: str) -> None:
861 if entity_n3.startswith("<") and entity_n3.endswith(">"):
862 entity_uris = self._expand_entities_with_merges({entity_n3[1:-1]})
863 for entity_uri in entity_uris:
864 if entity_uri not in self.reconstructed_entities:
865 self.reconstructed_entities.add(entity_uri)
866 result = self._reconstruct_entity_state(entity_uri)
867 if result is not None:
868 self._merge_entity_result(entity_uri, result)
870 def _reconstruct_entity_state(self, entity_uri: str) -> dict | None:
871 agnostic_entity = AgnosticEntity(
872 entity_uri,
873 config=self.config,
874 include_related_objects=False,
875 include_merged_entities=False,
876 include_reverse_relations=False,
877 )
878 if self.on_time:
879 entity_graphs, _, _ = agnostic_entity.get_state_at_time(
880 time=self.on_time, include_prov_metadata=False
881 )
882 return entity_graphs
883 entity_history = agnostic_entity.get_history(include_prov_metadata=False)
884 return entity_history[0]
886 def _merge_entity_result(self, entity_uri: str, entity_graphs: dict) -> None:
887 if self.on_time:
888 if entity_graphs:
889 for relevant_timestamp, quad_set in entity_graphs.items():
890 self.relevant_entities_graphs.setdefault(entity_uri, {})[
891 relevant_timestamp
892 ] = quad_set
893 elif entity_graphs.get(entity_uri):
894 self.relevant_entities_graphs.update(entity_graphs)
896 def _get_present_entities(self, triple: tuple) -> set[str]:
897 entities = set()
898 for expanded_triple in self._expand_triple_aliases(triple):
899 entities.update(self._get_present_entities_exact(expanded_triple))
900 return entities
902 def _get_present_entities_exact(self, triple: tuple) -> set[str]:
903 variables = [el for el in triple if el.startswith("?")]
904 if self.config["dataset"]["is_quadstore"]:
905 query = (
906 f"SELECT {' '.join(variables)} WHERE "
907 f"{{GRAPH ?_g {{{triple[0]} {triple[1]} {triple[2]}}} "
908 "FILTER(!CONTAINS(STR(?_g), '/prov/'))}"
909 )
910 else:
911 query = (
912 f"SELECT {' '.join(variables)} WHERE "
913 f"{{{triple[0]} {triple[1]} {triple[2]}}}"
914 )
915 results = Sparql(query, self.config).run_select_query()
916 bindings = results["results"]["bindings"]
917 var_name = triple[0][1:]
918 return {
919 b[var_name]["value"]
920 for b in bindings
921 if var_name in b and b[var_name]["type"] == "uri"
922 }
924 def _is_a_new_triple(self, triple: tuple, triples_checked: set) -> bool:
925 constants = _pattern_constants(triple)
926 for triple_checked in triples_checked:
927 if not constants.difference(_pattern_constants(triple_checked)):
928 return False
929 return True
931 def _get_query_to_update_queries(self, triple: tuple) -> str:
932 if self._use_qlever_index():
933 associations = "\n".join(
934 f"?snapshot <{QLEVER_HAS_WORD}> {Literal(token).n3()}."
935 for token in sorted(qlever_search_tokens(triple))
936 )
937 return f"""
938 SELECT ?updateQuery WHERE {{
939 ?snapshot <{ProvEntity.iri_has_update_query}> ?updateQuery.
940 {associations}
941 }}
942 """
943 if self.fuseki_full_text_search:
944 phrases = _fuseki_search_phrases(triple)
945 query_obj = '\\" AND \\"'.join(
946 _escape_search_term(phrase, '"') for phrase in phrases
947 )
948 return f"""
949 PREFIX text: <http://jena.apache.org/text#>
950 SELECT ?updateQuery WHERE {{
951 ?se text:query ("\\"{query_obj}\\"" {_FUSEKI_TEXT_SEARCH_LIMIT});
952 <{ProvEntity.iri_has_update_query}> ?updateQuery.
953 }}
954 """
955 return self.get_full_text_search(_pattern_search_terms(triple))
957 def _use_qlever_index(self) -> bool:
958 return (
959 self.qlever_full_text_search
960 and bool(self.config["provenance"]["triplestore_urls"])
961 and not self.config["provenance"]["file_paths"]
962 )
964 def get_full_text_search(self, terms: set) -> str:
965 if not terms:
966 query_to_identify = f"""
967 SELECT ?updateQuery
968 WHERE {{
969 ?snapshot <{ProvEntity.iri_has_update_query}> ?updateQuery.
970 }}
971 """
972 elif self.blazegraph_full_text_search:
973 query_obj = " ".join(_escape_search_term(term, '"') for term in terms)
974 query_to_identify = f"""
975 PREFIX bds: <http://www.bigdata.com/rdf/search#>
976 SELECT ?updateQuery
977 WHERE {{
978 ?snapshot <{ProvEntity.iri_has_update_query}> ?updateQuery.
979 ?updateQuery bds:search "{query_obj}";
980 bds:matchAllTerms 'true'.
981 }}
982 """
983 elif self.fuseki_full_text_search:
984 # The angle brackets keep the IRI a single token under the
985 # whitespace tokenizer the index requires, so a namespace root such
986 # as <http://www.w3.org/> no longer matches every IRI below it.
987 query_obj = '\\" AND \\"'.join(
988 _escape_search_term(f"<{term}>", '"') for term in terms
989 )
990 query_to_identify = f"""
991 PREFIX text: <http://jena.apache.org/text#>
992 SELECT ?updateQuery WHERE {{
993 ?se text:query ("\\"{query_obj}\\"" {_FUSEKI_TEXT_SEARCH_LIMIT});
994 <{ProvEntity.iri_has_update_query}> ?updateQuery.
995 }}
996 """
997 elif self.virtuoso_full_text_search:
998 query_obj = "' AND '".join(
999 _escape_search_term(term, '"', "'") for term in terms
1000 )
1001 query_to_identify = f"""
1002 PREFIX bif: <bif:>
1003 SELECT ?updateQuery
1004 WHERE {{
1005 ?snapshot <{ProvEntity.iri_has_update_query}> ?updateQuery.
1006 ?updateQuery bif:contains "'{query_obj}'".
1007 }}
1008 """
1009 elif self.graphdb_connector_name:
1010 quote = '"'
1011 con_queries = (
1012 f"con:query '{quote}"
1013 + f"{quote}'; con:query '{quote}".join(
1014 _escape_search_term(term, "'", '"') for term in terms
1015 )
1016 + f"{quote}'"
1017 )
1018 query_to_identify = f"""
1019 PREFIX con: <http://www.ontotext.com/connectors/lucene#>
1020 PREFIX con-inst: <http://www.ontotext.com/connectors/lucene/instance#>
1021 SELECT ?updateQuery
1022 WHERE {{
1023 ?snapshot <{ProvEntity.iri_has_update_query}> ?updateQuery.
1024 [] a con-inst:{self.graphdb_connector_name};
1025 {con_queries};
1026 con:entities ?snapshot.
1027 }}
1028 """
1029 else:
1030 escaped = [_escape_search_term(term, "'") for term in terms]
1031 filters = ").".join(
1032 f"FILTER CONTAINS (?updateQuery, '{term}'" for term in escaped
1033 )
1034 query_to_identify = f"""
1035 SELECT ?updateQuery
1036 WHERE {{
1037 ?snapshot <{ProvEntity.iri_has_update_query}> ?updateQuery.
1038 {filters}).
1039 }}
1040 """
1041 return query_to_identify
1043 def _find_entity_uris_in_update_queries(self, triple: tuple, entities: set) -> None:
1044 for expanded_triple in self._expand_triple_aliases(triple):
1045 self._find_entity_uris_in_update_queries_exact(expanded_triple, entities)
1047 def _find_entity_uris_in_update_queries_exact(
1048 self, triple: tuple, entities: set
1049 ) -> None:
1050 if not any(
1051 [
1052 self.blazegraph_full_text_search,
1053 self.fuseki_full_text_search,
1054 self.virtuoso_full_text_search,
1055 (self._use_qlever_index() and _pattern_search_terms(triple)),
1056 self.graphdb_connector_name,
1057 ]
1058 ):
1059 terms = _pattern_search_terms(triple)
1060 escaped = [_escape_search_term(term, "'") for term in terms]
1061 filter_clauses = "\n".join(
1062 f"FILTER CONTAINS (?updateQuery, '{term}')." for term in escaped
1063 )
1064 query = f"""
1065 SELECT ?entity ?updateQuery WHERE {{
1066 ?snapshot <{ProvEntity.iri_specialization_of}> ?entity;
1067 <{ProvEntity.iri_has_update_query}> ?updateQuery.
1068 {filter_clauses}
1069 }}
1070 """
1071 results = Sparql(query, self.config).run_select_query()
1072 for binding in results["results"]["bindings"]:
1073 matching_quads = _matching_update_quads(
1074 binding["updateQuery"]["value"], triple
1075 )
1076 if matching_quads:
1077 entities.add(binding["entity"]["value"])
1078 return
1079 query_to_identify = self._get_query_to_update_queries(triple)
1080 results = Sparql(query_to_identify, self.config).run_select_query()
1081 bindings = results["results"]["bindings"]
1082 if self.fuseki_full_text_search and len(bindings) >= _FUSEKI_TEXT_SEARCH_LIMIT:
1083 terms = ", ".join(sorted(_pattern_search_terms(triple)))
1084 msg = (
1085 f"The full-text search for {terms} returned "
1086 f"{_FUSEKI_TEXT_SEARCH_LIMIT} update queries, which is the "
1087 "limit, so the rest were dropped and the entities behind them "
1088 "would be missing from the answer."
1089 )
1090 raise ValueError(msg)
1091 for binding in bindings:
1092 for quad in _matching_update_quads(binding["updateQuery"]["value"], triple):
1093 subject = quad[0].removeprefix("<").removesuffix(">")
1094 if subject.startswith("_:"):
1095 continue
1096 entities.add(subject)
1098 def _find_entities_in_update_queries(
1099 self, triple: tuple, present_entities: set | None = None
1100 ):
1101 if present_entities is None:
1102 present_entities = set()
1103 relevant_entities_found = present_entities
1104 self._find_entity_uris_in_update_queries(triple, relevant_entities_found)
1105 if relevant_entities_found:
1106 relevant_entities_found = self._expand_entities_with_merges(
1107 relevant_entities_found
1108 )
1109 args_list = [
1110 (entity_uri, self.config, self.on_time)
1111 for entity_uri in relevant_entities_found
1112 ]
1113 for result in _run_in_parallel(_reconstruct_entity_worker, args_list):
1114 if result is not None:
1115 entity, entity_graphs = result
1116 self.reconstructed_entities.add(entity)
1117 self._merge_entity_result(entity, entity_graphs)
1119 def _term_matches(self, expected: str, actual: str, index: int) -> bool:
1120 if expected == actual:
1121 return True
1122 if (
1123 not self.merge_aware
1124 or index not in (0, _OBJECT_POS)
1125 or not expected.startswith("<")
1126 or not expected.endswith(">")
1127 or not actual.startswith("<")
1128 or not actual.endswith(">")
1129 ):
1130 return False
1131 return actual[1:-1] in self._entity_aliases(expected[1:-1])
1133 def _solve_variables(self) -> None:
1134 self.vars_to_explicit_by_time = {}
1135 self._get_vars_to_explicit_by_time()
1136 while self._there_are_variables():
1137 solved_variables = self._explicit_solvable_variables()
1138 self._align_snapshots()
1139 if not solved_variables:
1140 return
1141 self._update_vars_to_explicit(solved_variables)
1142 self._get_vars_to_explicit_by_time()
1144 def _there_are_variables(self) -> bool:
1145 for triples in self.vars_to_explicit_by_time.values():
1146 for triple in triples:
1147 if any(el.startswith("?") for el in triple):
1148 return True
1149 return False
1151 def _explicit_solvable_variables(self) -> dict:
1152 explicit_triples: dict[str, dict[str, set]] = {}
1153 for se, triples in self.vars_to_explicit_by_time.items():
1154 for triple in triples:
1155 variables = [el for el in triple if el.startswith("?")]
1156 if len(variables) == 1:
1157 variable = variables[0]
1158 variable_index = triple.index(variable)
1159 if variable_index == _OBJECT_POS:
1160 subject_terms = {triple[0]}
1161 if triple[0].startswith("<") and triple[0].endswith(">"):
1162 subject_terms = {
1163 f"<{uri}>"
1164 for uri in self._entity_aliases(triple[0][1:-1])
1165 }
1166 quads_by_subject = self._relevant_graphs_by_subject[se]
1167 query_results = [
1168 (triple[0], triple[1], q[2])
1169 for subject in subject_terms
1170 if subject in quads_by_subject
1171 for q in quads_by_subject[subject]
1172 if q[1] == triple[1]
1173 ]
1174 for row in query_results:
1175 explicit_triples.setdefault(se, {})
1176 explicit_triples[se].setdefault(variable, set())
1177 explicit_triples[se][variable].add(row)
1178 discovered_entities = {
1179 row[2][1:-1]
1180 for row in query_results
1181 if row[2].startswith("<") and row[2].endswith(">")
1182 }
1183 discovered_entities = self._expand_entities_with_merges(
1184 discovered_entities
1185 ).difference(self.reconstructed_entities)
1186 args_list = [
1187 (entity_uri, self.config, self.on_time)
1188 for entity_uri in discovered_entities
1189 ]
1190 for result_data in _run_in_parallel(
1191 _reconstruct_entity_worker, args_list
1192 ):
1193 if result_data is not None:
1194 entity, entity_graphs = result_data
1195 self.reconstructed_entities.add(entity)
1196 self._merge_entity_result(entity, entity_graphs)
1197 return explicit_triples
1199 def _align_snapshots(self) -> None:
1200 for snapshots in self.relevant_entities_graphs.values():
1201 for snapshot, quad_set in snapshots.items():
1202 if snapshot in self.relevant_graphs:
1203 self.relevant_graphs[snapshot].update(quad_set)
1204 else:
1205 self.relevant_graphs[snapshot] = set(quad_set)
1206 self._relevant_graphs_by_subject = {
1207 timestamp: _index_quads_by_subject(quad_set)
1208 for timestamp, quad_set in self.relevant_graphs.items()
1209 }
1210 if len(self.relevant_graphs) <= 1:
1211 return
1212 ordered_data = sorted(
1213 self.relevant_graphs.items(),
1214 key=lambda x: _parse_datetime(x[0]),
1215 )
1216 for index in range(1, len(ordered_data)):
1217 previous_se = ordered_data[index - 1][0]
1218 se = ordered_data[index][0]
1219 previous_by_subject = self._relevant_graphs_by_subject[previous_se]
1220 current_by_subject = self._relevant_graphs_by_subject[se]
1221 for subject_n3, subject_quads in previous_by_subject.items():
1222 subject_uri = (
1223 subject_n3[1:-1] if subject_n3.startswith("<") else subject_n3
1224 )
1225 if (
1226 subject_n3 not in current_by_subject
1227 and subject_uri in self.relevant_entities_graphs
1228 and se not in self.relevant_entities_graphs[subject_uri]
1229 ):
1230 self.relevant_graphs[se].update(subject_quads)
1231 current_by_subject[subject_n3] = set(subject_quads)
1233 def _update_vars_to_explicit(self, solved_variables: dict):
1234 vars_to_explicit_by_time: dict = {}
1235 for se, triples in self.vars_to_explicit_by_time.items():
1236 vars_to_explicit_by_time.setdefault(se, set())
1237 new_triples = set()
1238 for triple in triples:
1239 if se in solved_variables:
1240 for solved_var, solved_triples in solved_variables[se].items():
1241 if solved_var in triple:
1242 for solved_triple in solved_triples:
1243 new_triple = None
1244 if (
1245 solved_triple[0] != triple[0]
1246 and solved_triple[1] == triple[1]
1247 ):
1248 continue
1249 if (
1250 solved_triple[0] == triple[0]
1251 and solved_triple[1] == triple[1]
1252 ):
1253 new_triple = solved_triple
1254 else:
1255 new_triple = (
1256 solved_triple[2],
1257 triple[1],
1258 triple[2],
1259 )
1260 new_triples.add(new_triple)
1261 elif not any(el.startswith("?") for el in triple) or not any(
1262 var for var in solved_variables[se] if var in triple
1263 ):
1264 new_triples.add(triple)
1265 vars_to_explicit_by_time[se] = new_triples
1266 self.vars_to_explicit_by_time = vars_to_explicit_by_time
1268 def _get_vars_to_explicit_by_time(self) -> None:
1269 relevant_triples = None
1270 for se in self.relevant_graphs:
1271 if se not in self.vars_to_explicit_by_time:
1272 if relevant_triples is None:
1273 relevant_triples = set()
1274 for triple in self.triples:
1275 if any(
1276 el
1277 for el in triple
1278 if el.startswith("?")
1279 and not self._is_a_dead_end(el, triple)
1280 ) and not self._is_isolated(triple):
1281 relevant_triples.add(triple)
1282 self.vars_to_explicit_by_time[se] = set(relevant_triples)
1284 def _is_a_dead_end(self, el: str, triple: tuple) -> bool:
1285 return (
1286 el.startswith("?")
1287 and triple.index(el) == _OBJECT_POS
1288 and not any(t for t in self.triples if el in t if t.index(el) == 0)
1289 )
1292class VersionQuery(AgnosticQuery):
1293 """Time-travel queries, both on a single version and all versions of the dataset.
1295 :param query: The SPARQL query string.
1296 :type query: str
1297 :param on_time: If you want to query a specific version, specify the time
1298 interval here. The format is (START, END). If one of the two values is None,
1299 only the other is considered. Dates must be in ISO 8601 format.
1300 :type on_time: Tuple[Union[str, None]], optional
1301 :param merge_aware: Follow entity histories connected by merges.
1302 :type merge_aware: bool, optional
1303 :param include_prov_metadata: Return snapshot metadata with the query results.
1304 :type include_prov_metadata: bool, optional
1305 :param config_path: The path to the configuration file.
1306 :type config_path: str, optional
1307 """
1309 def __init__(
1310 self,
1311 query: str,
1312 on_time: tuple[str | None, str | None] | None = None,
1313 *,
1314 merge_aware: bool = False,
1315 include_prov_metadata: bool = False,
1316 config_path: str = CONFIG_PATH,
1317 config_dict: dict | None = None,
1318 ):
1319 self._streaming_results: dict[str, list[dict]] = {}
1320 super().__init__(
1321 query,
1322 on_time,
1323 merge_aware=merge_aware,
1324 include_prov_metadata=include_prov_metadata,
1325 config_path=config_path,
1326 config_dict=config_dict,
1327 )
1329 def _rebuild_relevant_graphs(self) -> None:
1330 self.triples = self._process_query()
1331 if self.on_time is not None:
1332 if (
1333 len(self.triples) == 1
1334 and self._is_isolated(self.triples[0])
1335 and not self.merge_aware
1336 ):
1337 self._rebuild_vm_batch(self.on_time)
1338 return
1339 super()._rebuild_relevant_graphs()
1340 return
1341 if not all(self._is_isolated(t) for t in self.triples):
1342 super()._rebuild_relevant_graphs()
1343 self._streaming_results = {
1344 str(convert_to_datetime(ts, stringify=True)): self._extract_bindings(g)
1345 for ts, g in self.relevant_graphs.items()
1346 }
1347 return
1348 self._rebuild_streaming()
1350 def _discover_entities_parallel(self, triple: tuple) -> set[str]:
1351 fut_present = _IO_EXECUTOR.submit(self._get_present_entities, triple)
1352 entities_set: set = set()
1353 fut_prov = _IO_EXECUTOR.submit(
1354 self._find_entity_uris_in_update_queries, triple, entities_set
1355 )
1356 present_entities = fut_present.result()
1357 fut_prov.result()
1358 all_entities = set(present_entities)
1359 all_entities.update(entities_set)
1360 return all_entities
1362 def _rebuild_vm_batch(self, on_time: tuple[str | None, str | None]) -> None:
1363 triple = self.triples[0]
1364 all_entity_strs = self._discover_entities_parallel(triple)
1365 all_entity_strs = self._expand_entities_with_merges(all_entity_strs)
1366 if not all_entity_strs:
1367 return
1368 self.reconstructed_entities.update(all_entity_strs)
1369 fut_prov = _IO_EXECUTOR.submit(
1370 _batch_query_provenance_snapshots,
1371 all_entity_strs,
1372 self.config,
1373 on_time[0],
1374 )
1375 fut_data = _IO_EXECUTOR.submit(
1376 _batch_query_dataset_triples,
1377 all_entity_strs,
1378 self.config,
1379 is_virtuoso=self.virtuoso_full_text_search,
1380 triple=triple,
1381 )
1382 prov_data = fut_prov.result()
1383 dataset_data = fut_data.result()
1384 point_time = (
1385 str(convert_to_datetime(on_time[0], stringify=True))
1386 if on_time[0] is not None and on_time[0] == on_time[1]
1387 else None
1388 )
1389 entity_bindings: dict[str, dict[str, list[dict]]] = {}
1390 for entity_str in all_entity_strs:
1391 per_ts: dict[str, list[dict]] = {}
1392 for ts, quad_set in _reconstruct_at_time_as_sets(
1393 prov_data[entity_str],
1394 dataset_data[entity_str],
1395 on_time,
1396 triple,
1397 ):
1398 result_time = point_time if point_time is not None else ts
1399 per_ts[result_time] = _match_single_pattern(triple, quad_set)
1400 entity_bindings[entity_str] = per_ts
1401 self._streaming_results = _merge_entity_bindings(entity_bindings)
1403 def _extract_bindings(self, quads: set[tuple[str, ...]]) -> list[dict]:
1404 # Match the SPARQL query patterns against the quad set.
1405 # Phase 1: mandatory triples. Start with an empty binding and for each
1406 # pattern keep only the quads that are compatible with what was already
1407 # matched.
1408 bindings: list[dict[str, str]] = [{}]
1409 for pattern in self._mandatory_triples:
1410 new_bindings: list[dict[str, str]] = []
1411 for binding in bindings:
1412 for quad in quads:
1413 new_binding = self._try_match(pattern, quad, binding)
1414 if new_binding is not None:
1415 new_bindings.append(new_binding)
1416 bindings = new_bindings
1417 # Phase 2: OPTIONAL groups. Try to extend each binding, but if nothing
1418 # matches, keep the binding as-is (no data is lost).
1419 for opt_group in self._optional_groups:
1420 bindings = self._left_join(bindings, opt_group, quads)
1421 # Phase 3: project to SELECT variables and deduplicate.
1422 seen: set[frozenset] = set()
1423 result: list[dict] = []
1424 for b in bindings:
1425 projected_n3: dict[str, str] = {}
1426 for var in self._select_vars:
1427 key = "?" + var
1428 if key in b:
1429 projected_n3[var] = b[key]
1430 frozen = frozenset(projected_n3.items())
1431 if not self._distinct or frozen not in seen:
1432 seen.add(frozen)
1433 result.append(
1434 {var: _n3_to_binding(val) for var, val in projected_n3.items()}
1435 )
1436 return result
1438 def _left_join(
1439 self,
1440 bindings: list[dict],
1441 opt_triples: list[tuple],
1442 quads: set[tuple[str, ...]],
1443 ) -> list[dict]:
1444 # For each binding, try to add values from the optional patterns.
1445 # If a quad matches, the binding grows with new variables.
1446 # If nothing matches, the binding is kept unchanged.
1447 result: list[dict] = []
1448 for binding in bindings:
1449 extended: list[dict] = [dict(binding)]
1450 for pattern in opt_triples:
1451 new_extended: list[dict] = []
1452 for b in extended:
1453 matched = False
1454 for quad in quads:
1455 new_b = self._try_match(pattern, quad, b)
1456 if new_b is not None:
1457 new_extended.append(new_b)
1458 matched = True
1459 if not matched:
1460 new_extended.append(b)
1461 extended = new_extended
1462 result.extend(extended)
1463 return result
1465 def _try_match(self, pattern: tuple, quad: tuple, binding: dict) -> dict | None:
1466 # Check if a triple pattern (s, p, o) matches a quad.
1467 new_binding = dict(binding)
1468 for index, (expected, actual) in enumerate(
1469 zip(pattern[:3], quad[:3], strict=True)
1470 ):
1471 is_variable = expected.startswith("?")
1472 if is_variable and expected in new_binding:
1473 # Variable already bound: check consistency
1474 if new_binding[expected] != actual:
1475 return None
1476 elif is_variable:
1477 # New variable: bind it
1478 new_binding[expected] = actual
1479 elif not self._term_matches(expected, actual, index):
1480 return None
1481 return new_binding
1483 def _rebuild_streaming(self) -> None:
1484 triples_checked = set()
1485 all_entity_strs: set[str] = set()
1486 use_fast_path = (
1487 len(self.triples) == 1
1488 and self._is_isolated(self.triples[0])
1489 and not self.merge_aware
1490 )
1491 for triple in self.triples:
1492 if self._is_a_new_triple(triple, triples_checked):
1493 present_entities = self._get_present_entities(triple)
1494 prov_entities: set = set()
1495 self._find_entity_uris_in_update_queries(triple, prov_entities)
1496 all_entity_strs.update(present_entities)
1497 all_entity_strs.update(prov_entities)
1498 triples_checked.add(triple)
1499 if not all_entity_strs:
1500 self._streaming_results = {}
1501 return
1502 all_entity_strs = self._expand_entities_with_merges(all_entity_strs)
1503 self.reconstructed_entities.update(all_entity_strs)
1504 if use_fast_path:
1505 fut_prov = _IO_EXECUTOR.submit(
1506 _batch_query_provenance_snapshots, all_entity_strs, self.config
1507 )
1508 fut_data = _IO_EXECUTOR.submit(
1509 _batch_query_dataset_triples,
1510 all_entity_strs,
1511 self.config,
1512 is_virtuoso=self.virtuoso_full_text_search,
1513 triple=self.triples[0],
1514 )
1515 prov_data = fut_prov.result()
1516 dataset_data = fut_data.result()
1517 triple = self.triples[0]
1518 quad_filter = _quad_filter_for_pattern(triple)
1519 entity_bindings: dict[str, dict[str, list[dict]]] = {}
1520 for entity_str in all_entity_strs:
1521 per_ts: dict[str, list[dict]] = {}
1522 sorted_versions = sorted(
1523 (
1524 (snapshot["time"], snapshot["updateQuery"])
1525 for snapshot in prov_data[entity_str]
1526 ),
1527 key=lambda version: _parse_datetime(version[0]),
1528 reverse=True,
1529 )
1530 for ts, quad_set in _iter_working_states(
1531 sorted_versions,
1532 dataset_data[entity_str],
1533 quad_filter=quad_filter,
1534 ):
1535 per_ts[ts] = _match_single_pattern(triple, quad_set)
1536 entity_bindings[entity_str] = per_ts
1537 else:
1538 entity_bindings = {}
1539 for entity_str in all_entity_strs:
1540 ae = AgnosticEntity(
1541 entity_str,
1542 config=self.config,
1543 include_related_objects=False,
1544 include_merged_entities=False,
1545 include_reverse_relations=False,
1546 )
1547 per_ts = {}
1548 for ts, quad_set in ae.iter_versions():
1549 per_ts[ts] = self._extract_bindings(quad_set)
1550 entity_bindings[entity_str] = per_ts
1551 self._streaming_results = _merge_entity_bindings(entity_bindings)
1553 def run_agnostic_query(
1554 self,
1555 ) -> tuple[dict[str, list[dict]], dict | None, dict | None]:
1556 is_point_query = (
1557 self.on_time is not None
1558 and self.on_time[0] is not None
1559 and self.on_time[0] == self.on_time[1]
1560 )
1561 if self._streaming_results:
1562 agnostic_result = self._streaming_results
1563 elif is_point_query:
1564 point_graph: set[tuple[str, ...]] = set()
1565 for graph in self.relevant_graphs.values():
1566 point_graph.update(graph)
1567 point_interval = cast("tuple[str | None, str | None]", self.on_time)
1568 point_time = cast("str", point_interval[0])
1569 agnostic_result = {point_time: self._extract_bindings(point_graph)}
1570 elif self.on_time is None:
1571 agnostic_result = self._streaming_results
1572 else:
1573 agnostic_result = {}
1574 for timestamp, graph in self.relevant_graphs.items():
1575 normalized = str(convert_to_datetime(timestamp, stringify=True))
1576 agnostic_result[normalized] = self._extract_bindings(graph)
1577 provenance, other_provenance = self._load_provenance(
1578 self.reconstructed_entities
1579 )
1580 return agnostic_result, provenance, other_provenance
1583class DeltaQuery(AgnosticQuery):
1584 """Delta structured query over a temporal interval.
1586 The result contains the multiset difference between the solution mappings
1587 at the interval endpoints and the difference for each consecutive state.
1588 If ``on_time`` is ``None``, the interval spans the entire dataset history.
1590 :param query: A SPARQL query string.
1591 :type query: str
1592 :param on_time: The time interval in the format (START, END). If one of the
1593 two values is None, only the other is considered. If the interval is
1594 None, the entire dataset history is considered. Dates must be in ISO
1595 8601 format.
1596 :type on_time: Tuple[Union[str, None]], optional
1597 :param merge_aware: Follow entity histories connected by merges.
1598 :type merge_aware: bool, optional
1599 :param include_prov_metadata: Return snapshot metadata with the query results.
1600 :type include_prov_metadata: bool, optional
1601 :param config_path: The path to the configuration file.
1602 :type config_path: str, optional
1603 """
1605 def __init__(
1606 self,
1607 query: str,
1608 on_time: tuple[str | None, str | None] | None = None,
1609 *,
1610 merge_aware: bool = False,
1611 include_prov_metadata: bool = False,
1612 config_path: str = CONFIG_PATH,
1613 config_dict: dict | None = None,
1614 ):
1615 super().__init__(
1616 query=query,
1617 on_time=on_time,
1618 merge_aware=merge_aware,
1619 include_prov_metadata=include_prov_metadata,
1620 config_path=config_path,
1621 config_dict=config_dict,
1622 )
1624 def _rebuild_relevant_graphs(self) -> None:
1625 version_query = VersionQuery(
1626 self.query,
1627 on_time=self.on_time,
1628 merge_aware=self.merge_aware,
1629 include_prov_metadata=False,
1630 config_dict=self.config,
1631 )
1632 self._version_results, _, _ = version_query.run_agnostic_query()
1633 self.reconstructed_entities = version_query.reconstructed_entities
1634 self._merge_events = version_query._merge_events # noqa: SLF001
1635 self.triples = version_query.triples
1637 def _reported_merge_events(self) -> list[dict] | None:
1638 if not self.merge_aware:
1639 return None
1640 return [
1641 {
1642 "time": event["time"],
1643 "snapshot": event["snapshot"],
1644 "survivor": event["survivor"],
1645 "absorbed": list(event["absorbed"]),
1646 }
1647 for event in self._merge_events.values()
1648 ]
1650 def run_agnostic_query(self) -> tuple[dict, dict | None, dict | None]:
1651 result = _build_solution_delta(self._version_results, self.on_time)
1652 result["merges"] = self._reported_merge_events()
1653 provenance, other_provenance = self._load_provenance(
1654 self.reconstructed_entities
1655 )
1656 return result, provenance, other_provenance
1659def get_insert_query(graph_iri: str, data: set[tuple[str, ...]]) -> tuple[str, int]:
1660 if not data:
1661 return "", 0
1662 statements = "\n".join(f"{q[0]} {q[1]} {q[2]} ." for q in data)
1663 return f"INSERT DATA {{ GRAPH <{graph_iri}> {{ {statements} }} }}", len(data)