diff --git a/Dockerfile b/Dockerfile index ad81833..3e811ed 100644 --- a/Dockerfile +++ b/Dockerfile @@ -5,6 +5,11 @@ ARG COMUNICA_VERSION=5.3.0 # behind it. It provides native SPARQL over a .hdt artifact, so validation can # query the compressed representation directly instead of decoding it first. ARG COMUNICA_HDT_VERSION=5.0.1 +#: Native COTTAS querying. Replaces the pycottas/rdflib path, which pulled +#: triples through the Store API in Python and could not answer a genotype +#: query at benchmark scale. Ships a DuckDB addon: glibc base only, and the +#: prebuild is per-architecture. +ARG COMUNICA_COTTAS_VERSION=0.1.0 # QLever is an optional second SPARQL engine for validation. Its binaries are # copied from the upstream published image rather than built here: compiling # QLever needs a large C++ toolchain and would dominate this image's build. @@ -175,6 +180,7 @@ RUN python3 -m venv /opt/pycottas-venv \ pyshacl==0.30.1 ARG COMUNICA_HDT_VERSION +ARG COMUNICA_COTTAS_VERSION # The HDT engine compiles native bindings, so a toolchain is needed at install # time but not afterwards; it is purged in the same layer to keep it out of the @@ -184,9 +190,31 @@ RUN apt-get update \ && npm install --global \ "@comunica/query-sparql-file@${COMUNICA_VERSION}" \ "@comunica/query-sparql-hdt@${COMUNICA_HDT_VERSION}" \ + "@elias.crum/query-sparql-cottas@${COMUNICA_COTTAS_VERSION}" \ && apt-get purge -y --auto-remove build-essential \ && rm -rf /var/lib/apt/lists/* +# A DuckDB addon that cannot load is a runtime failure inside a benchmark cell +# hours later, so prove the whole path here: write a triple, build a .cottas +# with pycottas, and answer a query over it with the comunica engine. +# +# Checking the binary alone is not enough. `--version` prints its table and +# then exits 1, and the addon is ESM so `require()` cannot probe it either -- +# both look like failures when the engine is fine. Only a real query settles it, +# and it also covers rdf2cottas, which is the other half of this engine. +RUN set -eu; \ + printf ' .\n' \ + > /tmp/cottas-probe.nt; \ + /opt/pycottas-venv/bin/python -c \ + "import pycottas; pycottas.rdf2cottas('/tmp/cottas-probe.nt', '/tmp/cottas-probe.cottas')"; \ + out="$(comunica-sparql-cottas cottas@/tmp/cottas-probe.cottas \ + 'SELECT ?s WHERE { ?s ?p ?o }' 2>&1)"; \ + printf '%s' "$out" | grep -q 'example.org/s' \ + || { echo "cottas engine built the artifact but cannot query it:"; \ + printf '%s\n' "$out"; exit 1; }; \ + rm -f /tmp/cottas-probe.nt /tmp/cottas-probe.cottas; \ + echo "cottas engine verified end to end" + RUN mkdir -p /opt/rmlstreamer \ && curl -fsSL \ -o /opt/rmlstreamer/RMLStreamer-v${RMLSTREAMER_VERSION}-standalone.jar \ diff --git a/src/validation/validation_runner.py b/src/validation/validation_runner.py index c3962de..77a6fb8 100644 --- a/src/validation/validation_runner.py +++ b/src/validation/validation_runner.py @@ -1971,6 +1971,7 @@ def validate_ntriples(source: Path, results_dir: Path) -> dict[str, Any]: #: The HDT endpoint runs alongside the N-Triples one within a single run, so #: they must not contend for a port. DEFAULT_HDT_ENDPOINT_PORT = 7021 +DEFAULT_COTTAS_ENDPOINT_PORT = 7022 QLEVER_STATUS_FILE = Path("/opt/vcf-rdfizer/qlever-status.txt") @@ -2118,6 +2119,25 @@ def run_query_process( return 124, f"query exceeded {timeout}s" +def is_timeout_error(error: BaseException) -> bool: + """Whether an exception from an HTTP query is a timeout rather than a fault. + + urlopen(timeout=...) surfaces a timeout either directly or wrapped in a + URLError, and the two are not interchangeable at the call site. Telling them + apart is what lets a timeout be reported as exit 124 instead of as a generic + execution failure -- which is the difference between "this query is too slow + for this engine" and "this engine is broken". + """ + import socket + import urllib.error + + if isinstance(error, (TimeoutError, socket.timeout)): + return True + if isinstance(error, urllib.error.URLError): + return isinstance(error.reason, (TimeoutError, socket.timeout)) + return False + + def _terminate_process_group(process: subprocess.Popen) -> None: """Signal a process group, escalating to SIGKILL, and reap the leader.""" for signal_number, grace in ((signal.SIGTERM, 10), (signal.SIGKILL, 5)): @@ -2385,12 +2405,19 @@ def execute(self, query_id: str, query_path: Path) -> dict[str, Any]: self.endpoint, query_path.read_text(encoding="utf-8"), timeout=self.query_timeout, ) - except Exception as error: # noqa: BLE001 - reported as EXECUTION_FAILED + except Exception as error: # noqa: BLE001 - reported below stderr_path.write_text(str(error), encoding="utf-8") raw_path.write_bytes(b"") + # 124 is the timeout convention the query loop keys on; anything + # else here really is an execution failure. + timed_out = is_timeout_error(error) + message = ( + f"query exceeded {self.query_timeout}s" if timed_out else str(error) + ) return self._envelope( - query_id, query_path, returncode=1, started=started, - raw_path=raw_path, stderr_path=stderr_path, error=str(error), + query_id, query_path, returncode=124 if timed_out else 1, + started=started, raw_path=raw_path, stderr_path=stderr_path, + error=message, ) raw_path.write_bytes(payload) stderr_path.write_bytes(b"") @@ -2618,12 +2645,17 @@ def execute(self, query_id: str, query_path: Path) -> dict[str, Any]: payload = self._post( query_path.read_text(encoding="utf-8"), timeout=self.query_timeout ) - except Exception as error: # noqa: BLE001 - reported as EXECUTION_FAILED + except Exception as error: # noqa: BLE001 - reported below stderr_path.write_text(str(error), encoding="utf-8") raw_path.write_bytes(b"") + timed_out = is_timeout_error(error) + message = ( + f"query exceeded {self.query_timeout}s" if timed_out else str(error) + ) return self._envelope( - query_id, query_path, returncode=1, started=started, - raw_path=raw_path, stderr_path=stderr_path, error=str(error), + query_id, query_path, returncode=124 if timed_out else 1, + started=started, raw_path=raw_path, stderr_path=stderr_path, + error=message, ) raw_path.write_bytes(payload) stderr_path.write_bytes(b"") @@ -2754,63 +2786,67 @@ def describe(self) -> dict[str, Any]: } -class CottasEngine(NativeArtifactEngine): - """Query a .cottas artifact in place through pycottas's rdflib store. +class CottasEngine(ComunicaHttpEndpointMixin, NativeArtifactEngine): + """Query a .cottas artifact in place, from one load, over Comunica's endpoint. + + This used to go through pycottas's rdflib Store in-process. rdflib evaluates + SPARQL over a custom Store by pulling triples through the Store API in + Python, which is why that path was unusable rather than merely slow: + measured on a 39 MB artifact, opening the store took 3-15 ms but + `SELECT ?s ?p ?o LIMIT 1` took 12.7 s and COUNT(*) did not finish in two + minutes. 06_equivalence hung on one genotype query for 41 hours, and again + for 10 hours once bounded, because a bound stops a query without making the + engine able to answer it. + + @elias.crum/query-sparql-cottas answers the same artifact through DuckDB + over the Parquet, behind the same long-lived endpoint the comunica and HDT + engines already use -- so the graph is opened once and every query is + charged only for itself. - pycottas exposes ``COTTASStore``, an rdflib Store backed by the Parquet - artifact, so this runs in-process rather than shelling out. The validator - already runs under the interpreter that owns pycottas. + Building the artifact still goes through pycottas: this replaces querying, + not rdf2cottas. """ name = "cottas" artifact_format = "cottas" + endpoint_binary = "comunica-sparql-cottas-http" + endpoint_port_option = "cottas_port" + default_endpoint_port = DEFAULT_COTTAS_ENDPOINT_PORT + endpoint_missing_hint = ( + "so COTTAS cannot be queried natively. Rebuild the image, or validate " + "the COTTAS artifact by decoding it (--rdf file.cottas --engine comunica)" + ) + + def __init__(self, source: Path, *, raw_dir: Path, scratch: Path, options: dict[str, Any]): + super().__init__(source, raw_dir=raw_dir, scratch=scratch, options=options) + self._init_endpoint(options) def build_artifact(self, target: Path) -> None: import pycottas pycottas.rdf2cottas(str(self.source), str(target)) - def start(self) -> None: - try: - import pycottas # noqa: F401 - import rdflib # noqa: F401 - except ImportError as error: - raise RuntimeError( - f"pycottas and rdflib are required to query COTTAS natively: {error}" - ) from error - super().start() - - import pycottas - import rdflib - - self.graph = rdflib.Graph(store=pycottas.COTTASStore(str(self.artifact))) + def _endpoint_source_argument(self) -> str: + # Comunica needs the source type declared, exactly as for HDT: a bare + # path is treated as a link to dereference. The engine takes one local + # file -- no URLs, directories or globs. + return f"cottas@{self.artifact}" - def execute(self, query_id: str, query_path: Path) -> dict[str, Any]: - raw_path = self.raw_dir / f"{query_id}.sparql.json" - stderr_path = self.raw_dir / f"{query_id}.stderr.txt" - started = time.monotonic() - try: - result = self.graph.query(query_path.read_text(encoding="utf-8")) - raw_path.write_bytes(result.serialize(format="json")) - stderr_path.write_bytes(b"") - returncode, error = 0, None - except Exception as failure: # noqa: BLE001 - reported as EXECUTION_FAILED - raw_path.write_text("{}", encoding="utf-8") - stderr_path.write_text(str(failure), encoding="utf-8") - returncode, error = 1, str(failure) - return self._envelope( - query_id, query_path, returncode=returncode, started=started, - raw_path=raw_path, stderr_path=stderr_path, error=error, - ) + def start(self) -> None: + # Check the binary before building the artifact: rdf2cottas is the + # expensive step and there is no point paying it for an engine that + # cannot run. + self._require_endpoint_binary() + NativeArtifactEngine.start(self) + self._start_endpoint() - def stop(self) -> None: - graph = getattr(self, "graph", None) - if graph is not None: - try: - graph.close() - except Exception: # noqa: BLE001 - teardown must not mask a result - pass - self.graph = None + def describe(self) -> dict[str, Any]: + return { + **NativeArtifactEngine.describe(self), + **self.endpoint_describe(), + "mode": "native cottas query over DuckDB, no decode", + "index_orders": "spo only (sibling .posg/.ospg files are not built)", + } ENGINE_CLASSES = { diff --git a/test/test_validation_benchmark_unit.py b/test/test_validation_benchmark_unit.py index 5e2989a..2dcafa6 100644 --- a/test/test_validation_benchmark_unit.py +++ b/test/test_validation_benchmark_unit.py @@ -227,27 +227,46 @@ def test_hdt_without_the_native_binary_explains_the_decode_alternative(self): engine.start() self.assertIn("--engine comunica", str(raised.exception)) - def test_a_failing_cottas_query_is_reported_not_raised(self): - """One bad query must not abort the other engines in a benchmark run.""" + def test_cottas_queries_are_addressed_with_comunicas_typed_source_prefix(self): + """Same requirement as HDT: a bare path is treated as a link.""" with tempfile.TemporaryDirectory() as td: tmp_path = Path(td) - raw_dir = tmp_path / "raw" - raw_dir.mkdir() - query = tmp_path / "q.rq" - query.write_text("SELECT * WHERE { ?s ?p ?o }", encoding="utf-8") + (tmp_path / "raw").mkdir() engine = self._engine("cottas", tmp_path) - engine.graph = mock.Mock() - engine.graph.query.side_effect = RuntimeError("parquet is unreadable") - envelope = engine.execute("q01", query) - - self.assertEqual(envelope["status"], "EXECUTION_FAILED") - self.assertIn("parquet is unreadable", envelope["error"]) - self.assertEqual(json.loads(Path(envelope["rawResult"]).read_text()), {}) - self.assertIn( - "parquet is unreadable", - Path(envelope["stderr"]).read_text(encoding="utf-8"), + engine.artifact = tmp_path / "cohort.cottas" + self.assertEqual( + engine._endpoint_source_argument(), + f"cottas@{tmp_path / 'cohort.cottas'}", ) + def test_cottas_is_queried_over_an_endpoint_not_in_process(self): + """The in-process rdflib Store is what made this engine unusable. + + rdflib evaluates SPARQL over a custom Store by pulling triples through + the Store API in Python: 12.7 s for LIMIT 1 on a 39 MB artifact, and a + genotype query that ran 41 hours without finishing. Querying over the + same long-lived endpoint the comunica and HDT engines use is what makes + 06_equivalence possible at all, so it must not regress to in-process. + """ + source = Path(V.__file__).read_text(encoding="utf-8") + cottas = source[source.index("class CottasEngine"):] + cottas = cottas[: cottas.index("\nENGINE_CLASSES")] + self.assertIn("ComunicaHttpEndpointMixin", cottas) + self.assertIn("comunica-sparql-cottas-http", cottas) + self.assertNotIn("self.graph.query", cottas) + self.assertNotIn("COTTASStore", cottas) + # Building the artifact is still pycottas' job; only querying moved. + self.assertIn("rdf2cottas", cottas) + + def test_cottas_checks_the_binary_before_building_the_artifact(self): + """rdf2cottas is the expensive step; do not pay it for a dead engine.""" + with tempfile.TemporaryDirectory() as td: + engine = self._engine("cottas", Path(td)) + with mock.patch.object(V.shutil, "which", return_value=None): + with self.assertRaises(RuntimeError) as raised: + engine.start() + self.assertIn("--engine comunica", str(raised.exception)) + def test_an_unknown_engine_name_names_the_supported_ones(self): with tempfile.TemporaryDirectory() as td: with self.assertRaises(ValueError) as raised: diff --git a/test/test_validation_engines_unit.py b/test/test_validation_engines_unit.py index 45b6e4a..a27f02c 100644 --- a/test/test_validation_engines_unit.py +++ b/test/test_validation_engines_unit.py @@ -10,6 +10,7 @@ import json import os import subprocess +import sys import tempfile import time import unittest @@ -877,3 +878,82 @@ def test_position_datatype_preflight_accepts_the_xsd_integer_family(self): self.assertIn(accepted, accepted_list) for rejected in ("xsd:string", "xsd:decimal", "xsd:double"): self.assertNotIn(rejected, accepted_list) + + +class QueryTimeoutEnforcementTests(VerboseTestCase): + """A query that will not finish must be stopped and reported as a timeout. + + Regression: 06_equivalence hung twice on the same cottas query -- 41 h, then + 10 h with --validation-query-timeout 1800 and --validation-time-budget 14400 + on the command line. Neither bound could fire. CottasEngine queried + in-process through rdflib, so nothing could interrupt it, and no engine ever + produced exit 124, so the loop's timeout handling was unreachable in any + case. run_query_process -- which does the right thing -- existed and had no + callers at all. + """ + + def test_run_query_process_reports_124_and_kills_the_tree(self): + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + out, err = tmp / "out.json", tmp / "err.txt" + marker = tmp / "child_survived" + # A parent that forks a child outliving it, exactly the shape the + # helper exists for: killing the leader must not leave the child. + source = ( + "import subprocess, sys, time\n" + "subprocess.Popen([sys.executable, '-c'," + " \"import time,pathlib;time.sleep(8);" + f"pathlib.Path(r'{marker}').write_text('x')\"])\n" + "time.sleep(60)\n" + ) + started = time.monotonic() + code, error = V.run_query_process( + [sys.executable, "-c", source], + stdout_path=out, stderr_path=err, timeout=2, + ) + elapsed = time.monotonic() - started + self.assertEqual(code, 124) + self.assertIn("exceeded", error or "") + self.assertLess(elapsed, 30, "the timeout did not actually interrupt") + time.sleep(9) + self.assertFalse( + marker.exists(), + "the forked child outlived the group kill and kept running", + ) + + def test_cottas_is_not_queried_in_process(self): + """In-process querying is what made the hang unkillable. + + It is now answered over comunica's endpoint, where the mixin's timeout + applies like any other HTTP engine -- so this engine inherits the 124 + classification rather than needing its own kill path. + """ + source = Path(V.__file__).read_text(encoding="utf-8") + cottas = source[source.index("class CottasEngine"):] + cottas = cottas[: cottas.index("\nENGINE_CLASSES")] + self.assertIn("ComunicaHttpEndpointMixin", cottas) + self.assertNotIn( + "self.graph.query", cottas, + "CottasEngine is querying in-process again; a hung query cannot be " + "stopped that way, which is the bug this replaced", + ) + + def test_http_timeouts_are_classified_as_timeouts(self): + """Exit 1 for a timeout makes it a crash; the loop keys on 124.""" + import socket + import urllib.error + self.assertTrue(V.is_timeout_error(TimeoutError())) + self.assertTrue(V.is_timeout_error(socket.timeout())) + self.assertTrue( + V.is_timeout_error(urllib.error.URLError(socket.timeout())) + ) + self.assertFalse(V.is_timeout_error(ValueError("bad query"))) + self.assertFalse( + V.is_timeout_error(urllib.error.URLError(ConnectionRefusedError())) + ) + + def test_the_timeout_convention_has_a_producer(self): + """124 is what the query loop keys on, so something must emit it.""" + source = Path(V.__file__).read_text(encoding="utf-8") + self.assertIn("return 124", source) + self.assertIn("returncode=124", source)