From 53412a7d23a4aca7f1916c6999c827040f204f30 Mon Sep 17 00:00:00 2001 From: ecrum19 Date: Fri, 18 Sep 2026 10:10:36 +0200 Subject: [PATCH 1/3] Make a COTTAS query killable, and make a timeout report as one 06_equivalence hung twice on the same query: 41 hours, then 10 hours WITH --validation-query-timeout 1800 and --validation-time-budget 14400 on the command line. Neither bound could fire, for two independent reasons. CottasEngine queried in-process through rdflib, so nothing could interrupt it. The call sits inside pycottas and no timeout reaches it; the time budget is checked between queries, so it never gets a turn either. And no engine ever produced exit 124, which is what the query loop keys on. The HTTP engines caught every exception and returned 1, so a timeout was indistinguishable from a crash, and --stop-after-query-timeout was unreachable in any configuration. run_query_process -- which kills the whole process group and returns 124 -- existed in the module with no callers at all. So: each COTTAS query now runs in its own process through run_query_process, and both HTTP engines classify a timeout as 124 via a new is_timeout_error rather than lumping it in with genuine faults. Measured before choosing the design, on a 39 MB artifact: opening the store takes 3-15 ms, `LIMIT 1` takes 12.7 s, and COUNT(*) does not finish in two minutes. Opening is free, so a process per query costs nothing and the load-once property the other engines need does not apply here -- for COTTAS the artifact is the index. A long-lived worker with a query protocol would have been complexity bought with an assumption the measurement contradicts. The numbers also say the hang is not a q05 bug: rdflib evaluates SPARQL over a custom Store by pulling triples through the Store API in Python, so this is what COTTAS costs at this scale. The fix bounds it; it does not make it fast. Tests cover the mechanism rather than the report: that run_query_process returns 124 and that a forked grandchild does not outlive the group kill, that CottasEngine no longer queries in-process, that timeouts classify correctly, and that something in the module still produces the 124 the loop depends on. Co-Authored-By: Claude Opus 5 --- src/validation/validation_runner.py | 109 ++++++++++++++++++------- test/test_validation_benchmark_unit.py | 40 ++++++++- test/test_validation_engines_unit.py | 75 +++++++++++++++++ 3 files changed, 192 insertions(+), 32 deletions(-) diff --git a/src/validation/validation_runner.py b/src/validation/validation_runner.py index c3962de..71d9203 100644 --- a/src/validation/validation_runner.py +++ b/src/validation/validation_runner.py @@ -2118,6 +2118,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 +2404,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 +2644,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,12 +2785,39 @@ def describe(self) -> dict[str, Any]: } +#: Run one COTTAS query and write SPARQL Results JSON to stdout. +#: +#: This is a separate process on purpose. rdflib evaluates SPARQL over a custom +#: Store by pulling triples through the Store API in Python, so a query that +#: looks trivial can run for hours: measured on a 39 MB artifact, opening the +#: store takes 3-15 ms but `SELECT ?s WHERE { ?s ?p ?o } LIMIT 1` takes 12.7 s +#: and `COUNT(*)` does not finish in two minutes. In-process, none of that is +#: interruptible -- the call sits inside pycottas and no timeout can reach it. +#: Out of process, it is killable. +#: +#: Re-opening per query costs nothing, so this does not undo the load-once +#: property the other engines need: for COTTAS the artifact *is* the index. +COTTAS_QUERY_RUNNER = """ +import sys + +import pycottas +import rdflib + +artifact, query_path = sys.argv[1], sys.argv[2] +graph = rdflib.Graph(store=pycottas.COTTASStore(artifact)) +with open(query_path, encoding="utf-8") as handle: + query = handle.read() +sys.stdout.buffer.write(graph.query(query).serialize(format="json")) +""" + + class CottasEngine(NativeArtifactEngine): """Query a .cottas artifact in place through pycottas's rdflib store. 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. + artifact. Each query runs in its own process rather than in this one, so + ``--validation-query-timeout`` can actually stop it -- see + COTTAS_QUERY_RUNNER for why that matters here and nowhere else. """ name = "cottas" @@ -2780,37 +2838,32 @@ def start(self) -> None: ) from error super().start() - import pycottas - import rdflib - - self.graph = rdflib.Graph(store=pycottas.COTTASStore(str(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 + returncode, error = run_query_process( + [sys.executable, "-c", COTTAS_QUERY_RUNNER, + str(self.artifact), str(query_path)], + stdout_path=raw_path, + stderr_path=stderr_path, + timeout=self.query_timeout, + ) + if returncode != 0: + # Whatever reached stdout is a truncated serialization, not a + # result; leave valid JSON so a reader fails on the status rather + # than on a parse error. raw_path.write_text("{}", encoding="utf-8") - stderr_path.write_text(str(failure), encoding="utf-8") - returncode, error = 1, str(failure) + if error is None: + error = stderr_path.read_text(encoding="utf-8")[-2000:] or None return self._envelope( query_id, query_path, returncode=returncode, started=started, raw_path=raw_path, stderr_path=stderr_path, error=error, ) 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 + # Nothing to close: each query owned its own process and has exited. + return None ENGINE_CLASSES = { diff --git a/test/test_validation_benchmark_unit.py b/test/test_validation_benchmark_unit.py index 5e2989a..2d7bcdd 100644 --- a/test/test_validation_benchmark_unit.py +++ b/test/test_validation_benchmark_unit.py @@ -228,7 +228,12 @@ def test_hdt_without_the_native_binary_explains_the_decode_alternative(self): 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.""" + """One bad query must not abort the other engines in a benchmark run. + + The query now runs in its own process so a hung one can be killed, so + the failure arrives as a non-zero exit with the child's stderr rather + than as an exception from an in-process rdflib call. + """ with tempfile.TemporaryDirectory() as td: tmp_path = Path(td) raw_dir = tmp_path / "raw" @@ -236,9 +241,19 @@ def test_a_failing_cottas_query_is_reported_not_raised(self): query = tmp_path / "q.rq" query.write_text("SELECT * WHERE { ?s ?p ?o }", encoding="utf-8") engine = self._engine("cottas", tmp_path) - engine.graph = mock.Mock() - engine.graph.query.side_effect = RuntimeError("parquet is unreadable") - envelope = engine.execute("q01", query) + engine.artifact = tmp_path / "missing.cottas" + with mock.patch.object( + V, "run_query_process", + return_value=(1, None), + ) as runner: + # The engine must surface whatever the child wrote, so put + # something there the way a real failure would. + def _fail(command, *, stdout_path, stderr_path, timeout): + stderr_path.write_text("parquet is unreadable", encoding="utf-8") + stdout_path.write_bytes(b"") + return 1, None + runner.side_effect = _fail + envelope = engine.execute("q01", query) self.assertEqual(envelope["status"], "EXECUTION_FAILED") self.assertIn("parquet is unreadable", envelope["error"]) @@ -248,6 +263,23 @@ def test_a_failing_cottas_query_is_reported_not_raised(self): Path(envelope["stderr"]).read_text(encoding="utf-8"), ) + def test_a_timed_out_cottas_query_is_reported_as_a_timeout(self): + """Exit 124 is what the query loop keys on to skip the rest.""" + with tempfile.TemporaryDirectory() as td: + tmp_path = Path(td) + (tmp_path / "raw").mkdir() + query = tmp_path / "q.rq" + query.write_text("SELECT * WHERE { ?s ?p ?o }", encoding="utf-8") + engine = self._engine("cottas", tmp_path) + engine.artifact = tmp_path / "a.cottas" + with mock.patch.object( + V, "run_query_process", + return_value=(124, "query exceeded 1800s"), + ): + envelope = engine.execute("q05", query) + self.assertEqual(envelope["exitCode"], 124) + self.assertIn("exceeded", envelope["error"]) + 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..b783505 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,77 @@ 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_runs_each_query_in_its_own_process(self): + """In-process querying is what made the hang unkillable.""" + source = Path(V.__file__).read_text(encoding="utf-8") + cottas = source[source.index("class CottasEngine"):] + cottas = cottas[: cottas.index("\nENGINE_CLASSES")] + self.assertIn("run_query_process", 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) From c6caab2c16458a253472932ab9a41d4ff16ab527 Mon Sep 17 00:00:00 2001 From: ecrum19 Date: Mon, 21 Sep 2026 15:19:53 +0200 Subject: [PATCH 2/3] Query COTTAS through comunica instead of rdflib The pycottas path could not answer a genotype query at benchmark scale. rdflib evaluates SPARQL over a custom Store by pulling triples through the Store API in Python: on a 39 MB artifact, opening the store took 3-15 ms while `SELECT ?s ?p ?o LIMIT 1` took 12.7 s and COUNT(*) did not finish in two minutes. 06_equivalence hung for 41 hours, then for 10 more once bounded -- a timeout 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, and ships comunica-sparql-cottas-http. That means CottasEngine becomes a sibling of HdtEngine rather than a special case: the same ComunicaHttpEndpointMixin, one long-lived endpoint, the graph opened once, each query charged only for itself, and the mixin's existing timeout handling inherited rather than reimplemented. Building the artifact still goes through pycottas' rdf2cottas; only querying moved. The binary check runs before NativeArtifactEngine.start so a dead engine does not first pay for rdf2cottas, which is the expensive step. Indexing is SPO only for now. The engine finds sibling .posg/.ospg files by naming convention if they exist and prunes patterns accordingly; none are built here, so every pattern is answered from the one order. Discovery is silent, so this is recorded in describe() rather than left to be inferred. The image installs the package globally beside the other comunica engines, and the build now fails if the binary will not run -- the DuckDB addon is a native prebuild with no musl build and per-architecture artifacts, and that failure would otherwise surface inside a benchmark cell hours later. Verified against the published 0.1.0 inside the current image on the benchmark host: it installs and answers a query on a real test-1k.cottas under Node 22.22.1, one patch below the version the integration guide states. Co-Authored-By: Claude Opus 5 --- Dockerfile | 12 +++ src/validation/validation_runner.py | 111 +++++++++++-------------- test/test_validation_benchmark_unit.py | 79 ++++++++---------- test/test_validation_engines_unit.py | 11 ++- 4 files changed, 100 insertions(+), 113 deletions(-) diff --git a/Dockerfile b/Dockerfile index ad81833..c899203 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,15 @@ 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. Prove the binary works while the image is still being built. +RUN comunica-sparql-cottas --version >/dev/null 2>&1 \ + || { echo "comunica-sparql-cottas is installed but will not run"; exit 1; } + 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 71d9203..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") @@ -2785,85 +2786,67 @@ def describe(self) -> dict[str, Any]: } -#: Run one COTTAS query and write SPARQL Results JSON to stdout. -#: -#: This is a separate process on purpose. rdflib evaluates SPARQL over a custom -#: Store by pulling triples through the Store API in Python, so a query that -#: looks trivial can run for hours: measured on a 39 MB artifact, opening the -#: store takes 3-15 ms but `SELECT ?s WHERE { ?s ?p ?o } LIMIT 1` takes 12.7 s -#: and `COUNT(*)` does not finish in two minutes. In-process, none of that is -#: interruptible -- the call sits inside pycottas and no timeout can reach it. -#: Out of process, it is killable. -#: -#: Re-opening per query costs nothing, so this does not undo the load-once -#: property the other engines need: for COTTAS the artifact *is* the index. -COTTAS_QUERY_RUNNER = """ -import sys +class CottasEngine(ComunicaHttpEndpointMixin, NativeArtifactEngine): + """Query a .cottas artifact in place, from one load, over Comunica's endpoint. -import pycottas -import rdflib - -artifact, query_path = sys.argv[1], sys.argv[2] -graph = rdflib.Graph(store=pycottas.COTTASStore(artifact)) -with open(query_path, encoding="utf-8") as handle: - query = handle.read() -sys.stdout.buffer.write(graph.query(query).serialize(format="json")) -""" + 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. -class CottasEngine(NativeArtifactEngine): - """Query a .cottas artifact in place through pycottas's rdflib store. - - pycottas exposes ``COTTASStore``, an rdflib Store backed by the Parquet - artifact. Each query runs in its own process rather than in this one, so - ``--validation-query-timeout`` can actually stop it -- see - COTTAS_QUERY_RUNNER for why that matters here and nowhere else. + 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() + 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() - returncode, error = run_query_process( - [sys.executable, "-c", COTTAS_QUERY_RUNNER, - str(self.artifact), str(query_path)], - stdout_path=raw_path, - stderr_path=stderr_path, - timeout=self.query_timeout, - ) - if returncode != 0: - # Whatever reached stdout is a truncated serialization, not a - # result; leave valid JSON so a reader fails on the status rather - # than on a parse error. - raw_path.write_text("{}", encoding="utf-8") - if error is None: - error = stderr_path.read_text(encoding="utf-8")[-2000:] or None - 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: - # Nothing to close: each query owned its own process and has exited. - return 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 2d7bcdd..2dcafa6 100644 --- a/test/test_validation_benchmark_unit.py +++ b/test/test_validation_benchmark_unit.py @@ -227,58 +227,45 @@ 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. - - The query now runs in its own process so a hung one can be killed, so - the failure arrives as a non-zero exit with the child's stderr rather - than as an exception from an in-process rdflib call. - """ + 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.artifact = tmp_path / "missing.cottas" - with mock.patch.object( - V, "run_query_process", - return_value=(1, None), - ) as runner: - # The engine must surface whatever the child wrote, so put - # something there the way a real failure would. - def _fail(command, *, stdout_path, stderr_path, timeout): - stderr_path.write_text("parquet is unreadable", encoding="utf-8") - stdout_path.write_bytes(b"") - return 1, None - runner.side_effect = _fail - 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_a_timed_out_cottas_query_is_reported_as_a_timeout(self): - """Exit 124 is what the query loop keys on to skip the rest.""" + 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: - tmp_path = Path(td) - (tmp_path / "raw").mkdir() - query = tmp_path / "q.rq" - query.write_text("SELECT * WHERE { ?s ?p ?o }", encoding="utf-8") - engine = self._engine("cottas", tmp_path) - engine.artifact = tmp_path / "a.cottas" - with mock.patch.object( - V, "run_query_process", - return_value=(124, "query exceeded 1800s"), - ): - envelope = engine.execute("q05", query) - self.assertEqual(envelope["exitCode"], 124) - self.assertIn("exceeded", envelope["error"]) + 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: diff --git a/test/test_validation_engines_unit.py b/test/test_validation_engines_unit.py index b783505..a27f02c 100644 --- a/test/test_validation_engines_unit.py +++ b/test/test_validation_engines_unit.py @@ -921,12 +921,17 @@ def test_run_query_process_reports_124_and_kills_the_tree(self): "the forked child outlived the group kill and kept running", ) - def test_cottas_runs_each_query_in_its_own_process(self): - """In-process querying is what made the hang unkillable.""" + 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("run_query_process", cottas) + self.assertIn("ComunicaHttpEndpointMixin", cottas) self.assertNotIn( "self.graph.query", cottas, "CottasEngine is querying in-process again; a hung query cannot be " From 3b2b0a1b78bf25516c967e5c4fae6aad9e7e190b Mon Sep 17 00:00:00 2001 From: ecrum19 Date: Mon, 21 Sep 2026 15:33:00 +0200 Subject: [PATCH 3/3] Verify the cottas engine with a real query at image-build time The first guard ran `comunica-sparql-cottas --version`, which prints its version table and then exits 1 -- so a working engine failed the build. The addon cannot be probed with require() either, because @duckdb/node-api is ESM. Both checks report a healthy engine as broken. Build a one-triple .cottas with pycottas and query it instead. That settles the question the guard is actually asking, and covers rdf2cottas as well, which is the half of this engine that did not move to comunica. Co-Authored-By: Claude Opus 5 --- Dockerfile | 22 +++++++++++++++++++--- 1 file changed, 19 insertions(+), 3 deletions(-) diff --git a/Dockerfile b/Dockerfile index c899203..3e811ed 100644 --- a/Dockerfile +++ b/Dockerfile @@ -195,9 +195,25 @@ RUN apt-get update \ && rm -rf /var/lib/apt/lists/* # A DuckDB addon that cannot load is a runtime failure inside a benchmark cell -# hours later. Prove the binary works while the image is still being built. -RUN comunica-sparql-cottas --version >/dev/null 2>&1 \ - || { echo "comunica-sparql-cottas is installed but will not run"; exit 1; } +# 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 \