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