Merge pull request 'feat(convert): integrate process metrics into JSON and SQLite converters' (#5) from dev_masha into master

Reviewed-on: #5

README.md

@@ -62,9 +62,9 @@ machine (no parallel work). Every task writes a completion marker
 | 03 CSV data generation | `generate_data.py` | `out/csv/` tree + `MANIFEST.csv` | implemented |
 | 04 Convert to JSON-LD | `convert_json.py` | `out/json/full/`, `out/json/hybrid/dataset.jsonld` | implemented |
 | 05 Convert to SQLite | `convert_sqlite.py` | `out/sqlite/tribo.db` | implemented |
-| 06 Convert to PostgreSQL | `convert_postgresql.py` | live `tribo` DB + `out/pg/tribo.dump` | planned |
-| 07 Convert to RDF | `convert_rdf.py` | `out/rdf/dataset.nt.gz`, oxigraph store | planned |
-| 08 Benchmark queries | `make_queries.py` | `out/bench/queries/`, `out/bench/expected/` | planned |
+| 06 Convert to PostgreSQL | `convert_postgresql.py` | live PostgreSQL DB (+ dump when pg_dump is available) | implemented |
+| 07 Convert to RDF | `convert_rdf.py` | `out/rdf/dataset.nt.gz`, `dataset.ttl`, oxigraph store | implemented |
+| 08 Benchmark queries | `make_queries.py` | `out/bench/queries/`, `out/bench/expected/` | implemented |
 | 09 Benchmark execution | `bench_runner.py` | `out/bench/results_raw.csv`, `results_median.csv` | planned |
 | 10 Extrapolation | `extrapolate.py` | `out/bench/extrapolation.csv`, `hardware_sizing.csv` | planned |
 | 11 Reporting | `report.py` | `out/report/REPORT.md`, charts, tables | planned |
@@ -147,11 +147,15 @@ docs/
 out/            ALL generated artifacts (git-ignored, reproducible):
   config/ csv/ json/ sqlite/ pg/ rdf/ bench/ report/ .done/
 common/               shared helpers (storage_sizes.csv contract, ...)
+  queries/            Q1-Q7 implementations for csv / json / rdf formats
 make_lab_config.py    task 01 entry script (one script per task, repo root)
 make_process_flow.py  task 02 entry script
 generate_data.py      task 03 entry script
 convert_json.py       task 04 entry script
 convert_sqlite.py     task 05 entry script
+convert_postgresql.py task 06 entry script (TRIBO_PG_DSN, TRIBO_PROM_URL)
+convert_rdf.py        task 07 entry script
+make_queries.py       task 08 entry script (canonical results need TRIBO_PG_DSN)
 requirements.txt      closed dependency list (docs/rules/code-python-style.md)
 ```
 

common/monitoring.py

@@ -0,0 +1,66 @@
+"""Remote host metrics via Prometheus / node_exporter (stdlib urllib only).
+
+The PostgreSQL instance is remote, so psutil cannot attribute its CPU/RAM
+(spec 09 assumed a local instance). The database host runs Prometheus;
+its base URL comes from the TRIBO_PROM_URL environment variable
+(e.g. http://192.168.10.73:9090). Unset = monitoring disabled.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import urllib.parse
+import urllib.request
+
+logger = logging.getLogger(__name__)
+
+QUERY_STEP_S = 5
+HTTP_TIMEOUT_S = 15
+# rate() window must span at least two scrapes (15 s default interval)
+CPU_QUERY = '100 * (1 - avg(rate(node_cpu_seconds_total{mode="idle"}[30s])))'
+RAM_QUERY = "node_memory_MemTotal_bytes - node_memory_MemAvailable_bytes"
+
+
+def get_prom_url() -> str:
+	return os.environ.get("TRIBO_PROM_URL", "")
+
+
+def query_range(base_url: str, promql: str, start: float, end: float) -> list[float]:
+	params = urllib.parse.urlencode({"query": promql, "start": start, "end": end, "step": QUERY_STEP_S})
+	url = base_url.rstrip("/") + "/api/v1/query_range?" + params
+	with urllib.request.urlopen(url, timeout=HTTP_TIMEOUT_S) as resp:
+		data = json.load(resp)
+	if data.get("status") != "success":
+		raise RuntimeError(f"prometheus query failed: {data.get('errorType')}: {data.get('error')}")
+	result = data["data"]["result"]
+	if not result:
+		return []
+	return [float(v) for _, v in result[0]["values"]]
+
+
+def host_window_metrics(base_url: str, start: float, end: float) -> dict | None:
+	"""CPU/RAM of the monitored host over [start, end] wall-clock seconds.
+
+	Returns None (with a warning) on any failure - monitoring is best-effort
+	and must never fail the owning task.
+	"""
+	try:
+		if end - start < 2 * QUERY_STEP_S:
+			end = start + 2 * QUERY_STEP_S
+		cpu = query_range(base_url, CPU_QUERY, start, end)
+		ram = query_range(base_url, RAM_QUERY, start, end)
+		if not cpu or not ram:
+			logger.warning("prometheus returned no samples for the window (expected condition: short window or scrape lag)")
+			return None
+		return {
+			"host_cpu_avg_pct": round(sum(cpu) / len(cpu), 1),
+			"host_cpu_max_pct": round(max(cpu), 1),
+			"host_ram_used_baseline_mb": round(ram[0] / 1048576),
+			"host_ram_used_peak_mb": round(max(ram) / 1048576),
+			"host_ram_used_delta_mb": round((max(ram) - ram[0]) / 1048576),
+		}
+	except Exception:
+		logger.warning("host metrics collection skipped (prometheus unreachable or query failed)", exc_info=True)
+		return None

common/pg.py

@@ -0,0 +1,16 @@
+"""PostgreSQL DSN access (docs/rules/code-config-yaml.md section 3).
+
+The DSN comes from the TRIBO_PG_DSN environment variable and is read ONLY
+here. It may embed a password: never log it, never hardcode it, never
+commit it - log host/database names at most.
+"""
+
+from __future__ import annotations
+
+import os
+
+DEFAULT_DSN = "postgresql://postgres@localhost:5432/tribo"
+
+
+def get_dsn() -> str:
+	return os.environ.get("TRIBO_PG_DSN", DEFAULT_DSN)

common/pipeline.py

@@ -10,6 +10,7 @@ import logging
 import sys
 from pathlib import Path
 
+import psutil
 import yaml
 
 logger = logging.getLogger(__name__)
@@ -57,3 +58,22 @@ def write_marker(out_root: Path, task_id: str, entries: dict) -> Path:
 def load_lab_config(out_root: Path) -> dict:
 	with open(out_root / "config" / "lab_config.yaml", encoding="utf-8") as fh:
 		return yaml.safe_load(fh)
+
+
+def process_metrics() -> dict:
+	"""Peak RSS and CPU time of the current process.
+
+	Every converter records its own build cost in its marker so the task 10
+	hardware-sizing model can extrapolate per-format RAM/CPU needs, not just
+	disk. peak_wset exists on Windows; the rss fallback covers Linux, where
+	the value is the current (not peak) RSS - good enough for streaming
+	converters whose RSS is flat by design.
+	"""
+	proc = psutil.Process()
+	mem = proc.memory_info()
+	peak = getattr(mem, "peak_wset", 0) or mem.rss
+	cpu = proc.cpu_times()
+	return {
+		"proc_peak_rss_mb": round(peak / 1048576, 1),
+		"proc_cpu_seconds": round(cpu.user + cpu.system, 1),
+	}

common/queries/__init__.py

@@ -0,0 +1,15 @@
+"""Benchmark query implementations for the file-based formats (spec 08).
+
+One module per format (csv_queries, json_queries, rdf_queries), each with
+functions q1..q7 returning canonical result rows (see common/results.py).
+The runnable wrappers under out/bench/queries/ are generated by
+make_queries.py and delegate here. SQLite / PostgreSQL implementations are
+plain SQL files written by make_queries.py.
+
+The relational formats answer Q2/Q3/Q4/Q7 from the precomputed
+track_summary (their idiomatic strength); the file formats derive the same
+values from raw cycles via common/track_summary.py - identical algorithm,
+identical results (bench-methodology section 5).
+"""
+
+Q1_TRACK_CODE = "B722-W2-C13-T2"  # fixed benchmark track, spec 08

common/queries/csv_queries.py

@@ -0,0 +1,139 @@
+"""Q1-Q7 over the raw CSV tree: pure python + csv module, streaming.
+
+Q1 is a direct path read (the honest CSV strength, spec 08); Q2-Q7 are
+deterministic directory walks with programmatic joins. No manifest
+validation here - queries measure retrieval, not integrity checking.
+"""
+
+from __future__ import annotations
+
+import csv
+from pathlib import Path
+
+from common.queries import Q1_TRACK_CODE
+from common.track_summary import summarize_track
+
+
+def _rows(path: Path) -> tuple[list[str], list[list[str]]]:
+	with open(path, newline="", encoding="ascii") as fh:
+		reader = csv.reader(fh)
+		header = [h.lower() for h in next(reader)]
+		return header, list(reader)
+
+
+def _dict_row(path: Path) -> dict[str, str]:
+	header, rows = _rows(path)
+	return dict(zip(header, rows[0]))
+
+
+def _track_path(csv_root: Path, track_code: str) -> Path:
+	batch, wafer, coupon, track = track_code.split("-")
+	return csv_root / f"batch_{batch}" / f"wafer_{wafer}" / f"coupon_{coupon}" / f"track_{track}"
+
+
+def _coupon_dirs(csv_root: Path):
+	for batch_dir in sorted(csv_root.glob("batch_*")):
+		batch_code = batch_dir.name[6:]
+		for wafer_dir in sorted(batch_dir.glob("wafer_*")):
+			for coupon_dir in sorted(wafer_dir.glob("coupon_*")):
+				yield batch_code, coupon_dir
+
+
+def _track_dirs(coupon_dir: Path) -> list[Path]:
+	return sorted(coupon_dir.glob("track_T*"))
+
+
+def _cofs(track_dir: Path) -> list[float]:
+	_, rows = _rows(track_dir / "cof_vs_cycle.csv")
+	return [float(r[1]) for r in rows]
+
+
+def _mean(values: list[float]) -> float:
+	total = 0.0
+	for v in values:
+		total += v
+	return total / len(values)
+
+
+def q1(csv_root: Path) -> list[tuple]:
+	_, rows = _rows(_track_path(csv_root, Q1_TRACK_CODE) / "cof_vs_cycle.csv")
+	return [(int(r[0]), float(r[1])) for r in rows]
+
+
+def q2(csv_root: Path) -> list[tuple]:
+	out = []
+	for _, coupon_dir in _coupon_dirs(csv_root):
+		info = _dict_row(coupon_dir / "coupon_info.csv")
+		if info["run_code"] == "RESERVE" or not 9.5 <= float(info["au_wtpct_mean"]) <= 10.5:
+			continue
+		tracks = _track_dirs(coupon_dir)
+		env = _dict_row(tracks[0] / "track_info.csv")["environment"]
+		cof_ss = _mean([summarize_track(_cofs(t))[0] for t in tracks])
+		out.append((info["coupon_code"], env, cof_ss))
+	return out
+
+
+def q3(csv_root: Path) -> list[tuple]:
+	out = []
+	for batch_code, coupon_dir in _coupon_dirs(csv_root):
+		info = _dict_row(coupon_dir / "coupon_info.csv")
+		if info["run_code"] == "RESERVE":
+			continue
+		header, rows = _rows(coupon_dir / "nanoindentation.csv")
+		h_col = header.index("hardness_gpa")
+		hardness = _mean([float(r[h_col]) for r in rows])
+		cof_ss = _mean([summarize_track(_cofs(t))[0] for t in _track_dirs(coupon_dir)])
+		out.append((info["coupon_code"], batch_code, hardness, cof_ss))
+	return out
+
+
+def q4(csv_root: Path) -> list[tuple]:
+	out = []
+	for _, coupon_dir in _coupon_dirs(csv_root):
+		for track_dir in _track_dirs(coupon_dir):
+			info = _dict_row(track_dir / "track_info.csv")
+			if info["environment"] != "dry_n2" or float(info["load_mn"]) != 100.0:
+				continue
+			cof_ss = summarize_track(_cofs(track_dir))[0]
+			if cof_ss > 0.20:
+				out.append((info["track_code"], cof_ss))
+	return out
+
+
+def q5(csv_root: Path) -> list[tuple]:
+	sums: dict[str, list[float]] = {}
+	for batch_code, coupon_dir in _coupon_dirs(csv_root):
+		for track_dir in _track_dirs(coupon_dir):
+			run_in = summarize_track(_cofs(track_dir))[2]
+			acc = sums.setdefault(batch_code, [0.0, 0.0])
+			acc[0] += run_in
+			acc[1] += 1
+	return [(batch, acc[0] / acc[1]) for batch, acc in sorted(sums.items())]
+
+
+def q6(csv_root: Path) -> list[tuple]:
+	out = []
+	for _, coupon_dir in _coupon_dirs(csv_root):
+		info = _dict_row(coupon_dir / "coupon_info.csv")
+		if info["run_code"] == "RESERVE":
+			continue
+		tracks = _track_dirs(coupon_dir)
+		load = float(_dict_row(tracks[0] / "track_info.csv")["load_mn"])
+		wear = _mean([float(_dict_row(t / "wear.csv")["wear_volume_um3"]) for t in tracks])
+		out.append((info["coupon_code"], load, wear))
+	return out
+
+
+def q7(csv_root: Path) -> list[tuple]:
+	groups: dict[tuple[str, int], list[float]] = {}
+	for _, coupon_dir in _coupon_dirs(csv_root):
+		for track_dir in _track_dirs(coupon_dir):
+			info = _dict_row(track_dir / "track_info.csv")
+			bucket = int(round(float(info["speed_mm_s"]) * float(info["load_mn"])))
+			cofs = _cofs(track_dir)
+			run_in = summarize_track(cofs)[2]
+			acc = groups.setdefault((info["environment"], bucket), [0.0, 0.0])
+			for cof in cofs[run_in:]:  # cycles strictly past run-in
+				acc[0] += cof
+				acc[1] += 1
+	return [(env, bucket, acc[0] / acc[1]) for (env, bucket), acc in sorted(groups.items())]

common/queries/json_queries.py

@@ -0,0 +1,169 @@
+"""Q1-Q7 over the JSON-LD FULL variant: ijson streaming parser (spec 08).
+
+Each coupon file is streamed event-by-event; scalar coupon fields appear
+before the bulk arrays in the serialization, so filtering queries (Q2) can
+abandon a non-matching file before parsing its megabytes of points.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import ijson
+
+from common.queries import Q1_TRACK_CODE
+from common.track_summary import summarize_track
+
+
+class CouponDoc:
+	__slots__ = ("coupon_code", "batch_code", "run_code", "au", "hardness", "tracks")
+
+	def __init__(self) -> None:
+		self.coupon_code = ""
+		self.batch_code = ""
+		self.run_code = ""
+		self.au = 0.0
+		self.hardness: list[float] = []
+		self.tracks: list[dict] = []
+
+
+def _scan(path: Path, au_range: tuple[float, float] | None = None) -> CouponDoc | None:
+	"""Stream one coupon file; with au_range set, bail out early on mismatch."""
+	doc = CouponDoc()
+	track: dict | None = None
+	with open(path, "rb") as fh:
+		for prefix, event, value in ijson.parse(fh, use_float=True):
+			if prefix == "coupon_code":
+				doc.coupon_code = value
+			elif prefix == "batch_code":
+				doc.batch_code = value
+			elif prefix == "run_code":
+				doc.run_code = value
+				if doc.run_code == "RESERVE":
+					return None
+			elif prefix == "au_wtpct_mean":
+				doc.au = value
+				if au_range and not au_range[0] <= value <= au_range[1]:
+					return None
+			elif prefix == "nanoindentation.indents.item.hardness_gpa":
+				doc.hardness.append(value)
+			elif prefix == "tracks.item" and event == "start_map":
+				track = {"cycles": [], "cofs": []}
+			elif track is not None:
+				if prefix == "tracks.item.track_code":
+					track["track_code"] = value
+				elif prefix == "tracks.item.environment":
+					track["environment"] = value
+				elif prefix == "tracks.item.load_mn":
+					track["load_mn"] = value
+				elif prefix == "tracks.item.speed_mm_s":
+					track["speed_mm_s"] = value
+				elif prefix == "tracks.item.cycles.item.cycle":
+					track["cycles"].append(value)
+				elif prefix == "tracks.item.cycles.item.cof":
+					track["cofs"].append(value)
+				elif prefix == "tracks.item.wear.wear_volume_um3":
+					track["wear"] = value
+				elif prefix == "tracks.item" and event == "end_map":
+					doc.tracks.append(track)
+					track = None
+	return doc
+
+
+def _files(json_root: Path) -> list[Path]:
+	return sorted((json_root / "full").glob("*.jsonld"))
+
+
+def _mean(values: list[float]) -> float:
+	total = 0.0
+	for v in values:
+		total += v
+	return total / len(values)
+
+
+def _cof_ss(doc: CouponDoc) -> float:
+	return _mean([summarize_track(t["cofs"])[0] for t in doc.tracks])
+
+
+def q1(json_root: Path) -> list[tuple]:
+	coupon_code = Q1_TRACK_CODE.rsplit("-", 1)[0]
+	doc = _scan(json_root / "full" / f"{coupon_code}.jsonld")
+	for track in doc.tracks:
+		if track["track_code"] == Q1_TRACK_CODE:
+			return [(int(c), cof) for c, cof in zip(track["cycles"], track["cofs"])]
+	raise LookupError(f"track {Q1_TRACK_CODE} not found")
+
+
+def q2(json_root: Path) -> list[tuple]:
+	out = []
+	for path in _files(json_root):
+		doc = _scan(path, au_range=(9.5, 10.5))
+		if doc is None:
+			continue
+		out.append((doc.coupon_code, doc.tracks[0]["environment"], _cof_ss(doc)))
+	return out
+
+
+def q3(json_root: Path) -> list[tuple]:
+	out = []
+	for path in _files(json_root):
+		doc = _scan(path)
+		if doc is None:
+			continue
+		out.append((doc.coupon_code, doc.batch_code, _mean(doc.hardness), _cof_ss(doc)))
+	return out
+
+
+def q4(json_root: Path) -> list[tuple]:
+	out = []
+	for path in _files(json_root):
+		doc = _scan(path)
+		if doc is None:
+			continue
+		for track in doc.tracks:
+			if track["environment"] != "dry_n2" or track["load_mn"] != 100.0:
+				continue
+			cof_ss = summarize_track(track["cofs"])[0]
+			if cof_ss > 0.20:
+				out.append((track["track_code"], cof_ss))
+	return out
+
+
+def q5(json_root: Path) -> list[tuple]:
+	sums: dict[str, list[float]] = {}
+	for path in _files(json_root):
+		doc = _scan(path)
+		if doc is None:
+			continue
+		acc = sums.setdefault(doc.batch_code, [0.0, 0.0])
+		for track in doc.tracks:
+			acc[0] += summarize_track(track["cofs"])[2]
+			acc[1] += 1
+	return [(batch, acc[0] / acc[1]) for batch, acc in sorted(sums.items())]
+
+
+def q6(json_root: Path) -> list[tuple]:
+	out = []
+	for path in _files(json_root):
+		doc = _scan(path)
+		if doc is None:
+			continue
+		load = doc.tracks[0]["load_mn"]
+		out.append((doc.coupon_code, load, _mean([t["wear"] for t in doc.tracks])))
+	return out
+
+
+def q7(json_root: Path) -> list[tuple]:
+	groups: dict[tuple[str, int], list[float]] = {}
+	for path in _files(json_root):
+		doc = _scan(path)
+		if doc is None:
+			continue
+		for track in doc.tracks:
+			bucket = int(round(track["speed_mm_s"] * track["load_mn"]))
+			run_in = summarize_track(track["cofs"])[2]
+			acc = groups.setdefault((track["environment"], bucket), [0.0, 0.0])
+			for cof in track["cofs"][run_in:]:
+				acc[0] += cof
+				acc[1] += 1
+	return [(env, bucket, acc[0] / acc[1]) for (env, bucket), acc in sorted(groups.items())]

common/queries/rdf_queries.py

@@ -0,0 +1,173 @@
+"""Q1-Q7 over the oxigraph triplestore: SPARQL retrieval (spec 08).
+
+The store holds raw data only (no precomputed summaries), so queries that
+need steady-state COF or run-in retrieve the raw cycles via SPARQL and
+derive the values with the shared algorithm (common/track_summary.py) -
+identical semantics to every other format. The SPARQL text of each query
+is exported to out/bench/queries/rdf/q<N>.sparql by make_queries.py.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from pyoxigraph import Store
+
+from common.queries import Q1_TRACK_CODE
+from common.track_summary import summarize_track
+
+PREFIX = "PREFIX tribo: <https://sandia.gov/ontology/tribology#>\n"
+TRACK_IRI = f"<https://sandia.gov/ontology/tribology/id/track/{Q1_TRACK_CODE}>"
+
+SPARQL = {
+	"q1": PREFIX + f"""SELECT ?cycle ?cof WHERE {{
+  {TRACK_IRI} tribo:hasCycle ?c .
+  ?c tribo:cycle ?cycle ; tribo:cof ?cof .
+}}""",
+	"q2": PREFIX + """SELECT ?ccode ?env ?track ?cycle ?cof WHERE {
+  ?coupon tribo:au_wtpct_mean ?au ; tribo:coupon_code ?ccode .
+  FILTER(?au >= 9.5 && ?au <= 10.5)
+  ?track tribo:partOf ?coupon ; tribo:environment ?env ; tribo:hasCycle ?c .
+  ?c tribo:cycle ?cycle ; tribo:cof ?cof .
+}""",
+	"q3_hardness": PREFIX + """SELECT ?ccode ?bcode (AVG(?h) AS ?hardness) WHERE {
+  ?coupon tribo:coupon_code ?ccode ; tribo:cutFrom ?wafer ; tribo:nanoindentation ?m .
+  ?wafer tribo:partOf ?batch .
+  ?batch tribo:batch_code ?bcode .
+  ?m tribo:hasIndent ?i .
+  ?i tribo:hardness_gpa ?h .
+} GROUP BY ?ccode ?bcode""",
+	"q3_cycles": PREFIX + """SELECT ?ccode ?track ?cycle ?cof WHERE {
+  ?track a tribo:FrictionTest ; tribo:partOf ?coupon ; tribo:hasCycle ?c .
+  ?coupon tribo:coupon_code ?ccode .
+  ?c tribo:cycle ?cycle ; tribo:cof ?cof .
+}""",
+	"q4": PREFIX + """SELECT ?tcode ?track ?cycle ?cof WHERE {
+  ?track tribo:environment "dry_n2" ; tribo:load_mn ?load ; tribo:track_code ?tcode ; tribo:hasCycle ?c .
+  FILTER(?load = 100)
+  ?c tribo:cycle ?cycle ; tribo:cof ?cof .
+}""",
+	"q5": PREFIX + """SELECT ?bcode ?track ?cycle ?cof WHERE {
+  ?track a tribo:FrictionTest ; tribo:partOf ?coupon ; tribo:hasCycle ?c .
+  ?coupon tribo:cutFrom ?wafer .
+  ?wafer tribo:partOf ?batch .
+  ?batch tribo:batch_code ?bcode .
+  ?c tribo:cycle ?cycle ; tribo:cof ?cof .
+}""",
+	"q6": PREFIX + """SELECT ?ccode ?load (AVG(?wv) AS ?wear) WHERE {
+  ?track tribo:partOf ?coupon ; tribo:load_mn ?load ; tribo:wear ?w .
+  ?w tribo:wear_volume_um3 ?wv .
+  ?coupon tribo:coupon_code ?ccode .
+} GROUP BY ?ccode ?load""",
+	"q7": PREFIX + """SELECT ?env ?speed ?load ?track ?cycle ?cof WHERE {
+  ?track tribo:environment ?env ; tribo:speed_mm_s ?speed ; tribo:load_mn ?load ; tribo:hasCycle ?c .
+  ?c tribo:cycle ?cycle ; tribo:cof ?cof .
+}""",
+}
+
+
+def _store(store_path: Path) -> Store:
+	try:
+		return Store.read_only(str(store_path))
+	except AttributeError:
+		return Store(str(store_path))
+
+
+def _track_cofs(solutions, track_var: str, keep: tuple[str, ...]) -> dict[str, dict]:
+	"""Group cycle solutions by track: ordered cof list + kept literals."""
+	tracks: dict[str, dict] = {}
+	for sol in solutions:
+		key = str(sol[track_var])
+		entry = tracks.get(key)
+		if entry is None:
+			entry = {"pairs": []}
+			for name in keep:
+				entry[name] = sol[name].value
+			tracks[key] = entry
+		entry["pairs"].append((int(sol["cycle"].value), float(sol["cof"].value)))
+	for entry in tracks.values():
+		entry["pairs"].sort()
+		entry["cofs"] = [cof for _, cof in entry["pairs"]]
+	return tracks
+
+
+def _mean(values: list[float]) -> float:
+	total = 0.0
+	for v in values:
+		total += v
+	return total / len(values)
+
+
+def q1(store_path: Path) -> list[tuple]:
+	store = _store(store_path)
+	return [(int(s["cycle"].value), float(s["cof"].value)) for s in store.query(SPARQL["q1"])]
+
+
+def q2(store_path: Path) -> list[tuple]:
+	store = _store(store_path)
+	tracks = _track_cofs(store.query(SPARQL["q2"]), "track", ("ccode", "env"))
+	coupons: dict[tuple[str, str], list[float]] = {}
+	for entry in sorted(tracks.values(), key=lambda e: e["ccode"]):
+		coupons.setdefault((entry["ccode"], entry["env"]), []).append(summarize_track(entry["cofs"])[0])
+	return [(code, env, _mean(values)) for (code, env), values in sorted(coupons.items())]
+
+
+def q3(store_path: Path) -> list[tuple]:
+	store = _store(store_path)
+	hardness = {
+		s["ccode"].value: (s["bcode"].value, float(s["hardness"].value))
+		for s in store.query(SPARQL["q3_hardness"])
+	}
+	tracks = _track_cofs(store.query(SPARQL["q3_cycles"]), "track", ("ccode",))
+	coupons: dict[str, list[float]] = {}
+	for entry in sorted(tracks.values(), key=lambda e: e["ccode"]):
+		coupons.setdefault(entry["ccode"], []).append(summarize_track(entry["cofs"])[0])
+	return [
+		(code, hardness[code][0], hardness[code][1], _mean(values))
+		for code, values in sorted(coupons.items())
+	]
+
+
+def q4(store_path: Path) -> list[tuple]:
+	store = _store(store_path)
+	tracks = _track_cofs(store.query(SPARQL["q4"]), "track", ("tcode",))
+	out = []
+	for entry in sorted(tracks.values(), key=lambda e: e["tcode"]):
+		cof_ss = summarize_track(entry["cofs"])[0]
+		if cof_ss > 0.20:
+			out.append((entry["tcode"], cof_ss))
+	return out
+
+
+def q5(store_path: Path) -> list[tuple]:
+	store = _store(store_path)
+	tracks = _track_cofs(store.query(SPARQL["q5"]), "track", ("bcode",))
+	sums: dict[str, list[float]] = {}
+	for entry in sorted(tracks.values(), key=lambda e: e["bcode"]):
+		acc = sums.setdefault(entry["bcode"], [0.0, 0.0])
+		acc[0] += summarize_track(entry["cofs"])[2]
+		acc[1] += 1
+	return [(batch, acc[0] / acc[1]) for batch, acc in sorted(sums.items())]
+
+
+def q6(store_path: Path) -> list[tuple]:
+	store = _store(store_path)
+	return [
+		(s["ccode"].value, float(s["load"].value), float(s["wear"].value))
+		for s in store.query(SPARQL["q6"])
+	]
+
+
+def q7(store_path: Path) -> list[tuple]:
+	store = _store(store_path)
+	tracks = _track_cofs(store.query(SPARQL["q7"]), "track", ("env", "speed", "load"))
+	groups: dict[tuple[str, int], list[float]] = {}
+	for key in sorted(tracks):
+		entry = tracks[key]
+		bucket = int(round(float(entry["speed"]) * float(entry["load"])))
+		run_in = summarize_track(entry["cofs"])[2]
+		acc = groups.setdefault((entry["env"], bucket), [0.0, 0.0])
+		for cof in entry["cofs"][run_in:]:
+			acc[0] += cof
+			acc[1] += 1
+	return [(env, bucket, acc[0] / acc[1]) for (env, bucket), acc in sorted(groups.items())]

common/relational.py

@@ -0,0 +1,161 @@
+"""Shared relational row stream for the SQLite and PostgreSQL converters.
+
+Walks the CSV corpus once (manifest-validated via CorpusReader) and yields
+(table, row_tuple) pairs in FK-dependency-safe order: a parent row is always
+yielded before any row that references it. Both engines load the SAME
+logical schema from this stream (docs/rules/db-sql-schema.md section 6).
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Iterator
+
+from common.corpus import CorpusReader
+from common.track_summary import summarize_track
+
+logger = logging.getLogger(__name__)
+
+# Dependency-safe order; consumers that buffer rows must flush in this order.
+TABLES = [
+	"instruments",
+	"batches",
+	"runs",
+	"wafers",
+	"coupons",
+	"tracks",
+	"simtra_profiles",
+	"xrf_points",
+	"profilometry_points",
+	"nanoindentation",
+	"afm",
+	"friction_cycles",
+	"friction_loop_points",
+	"wear",
+	"track_summary",
+]
+
+COLUMNS = {
+	"instruments": ("instrument_id", "instrument_code", "name", "role"),
+	"batches": ("batch_id", "batch_code", "pt_gun_tilt_deg", "au_gun_tilt_deg", "pt_power_w", "au_power_w", "pt_discharge_v", "au_discharge_v", "deposition_date"),
+	"runs": ("run_id", "run_code", "environment", "date", "plates", "operator"),
+	"wafers": ("wafer_id", "wafer_code", "batch_id", "wafer_index", "deposition_date", "coupons", "friction_coupons", "reserve_coupons"),
+	"coupons": ("coupon_id", "coupon_code", "wafer_id", "batch_id", "grid_row", "grid_col", "thickness_um", "ra_nm", "au_wtpct_mean", "run_id", "plate", "probe", "square"),
+	"tracks": ("track_id", "track_code", "coupon_id", "run_id", "environment", "load_mn", "stroke_mm", "speed_mm_s", "counterface_id", "started_at"),
+	"simtra_profiles": ("batch_id", "row_no", "angle_deg", "energy_ev", "pt_flux", "au_flux"),
+	"xrf_points": ("coupon_id", "grid_x", "grid_y", "pt_wtpct", "au_wtpct"),
+	"profilometry_points": ("coupon_id", "grid_x", "grid_y", "thickness_um"),
+	"nanoindentation": ("coupon_id", "indent_id", "x_um", "y_um", "hardness_gpa", "reduced_modulus_gpa", "max_load_mn"),
+	"afm": ("coupon_id", "ra_nm", "rq_nm", "image_file"),
+	"friction_cycles": ("track_id", "cycle", "cof"),
+	"friction_loop_points": ("track_id", "cycle", "pt", "position_um", "friction_force_mn"),
+	"wear": ("track_id", "wear_volume_um3", "k_archard", "sliding_distance_m"),
+	"track_summary": ("track_id", "cof_ss_mean", "cof_ss_std", "run_in_cycles"),
+}
+
+EXPECTED_COUNTS_KEYS = TABLES  # every table is count-validated after load
+
+
+def expected_counts(cfg: dict) -> dict[str, int]:
+	v = cfg["volumes"]
+	h = cfg["hierarchy"]
+	return {
+		"instruments": len(cfg["instruments"]),
+		"batches": h["batches"],
+		"runs": cfg["friction_assignment"]["runs"],
+		"wafers": h["batches"] * h["wafers_per_batch"],
+		"coupons": v["coupons_total"],
+		"tracks": v["tracks_total"],
+		"simtra_profiles": h["batches"] * v["simtra_rows_per_batch"],
+		"xrf_points": v["xrf_points_total"],
+		"profilometry_points": v["coupons_total"] * v["profilometry_points_per_coupon"],
+		"nanoindentation": v["coupons_total"] * v["nanoindentation_indents_per_coupon"],
+		"afm": v["coupons_total"],
+		"friction_cycles": v["cycle_rows_total"],
+		"friction_loop_points": v["loop_points_total"],
+		"wear": v["tracks_total"],
+		"track_summary": v["tracks_total"],
+	}
+
+
+def stream_rows(cfg: dict, reader: CorpusReader) -> Iterator[tuple[str, tuple]]:
+	batch_ids: dict[str, int] = {}
+	run_ids: dict[str, int] = {}
+
+	for i, inst in enumerate(cfg["instruments"], 1):
+		yield "instruments", (i, inst["instrument_id"], inst["name"], inst["role"])
+	for i, row in enumerate(reader.dicts("batches.csv"), 1):
+		batch_ids[row["batch_code"]] = i
+		yield "batches", (
+			i, row["batch_code"], row["pt_gun_tilt_deg"], row["au_gun_tilt_deg"],
+			row["pt_power_w"], row["au_power_w"], row["pt_discharge_v"],
+			row["au_discharge_v"], row["deposition_date"],
+		)
+	for i, row in enumerate(reader.dicts("runs.csv"), 1):
+		run_ids[row["run_code"]] = i
+		yield "runs", (i, row["run_code"], row["environment"], row["date"], row["plates"], row["operator"])
+	for code, batch_id in batch_ids.items():
+		for row_no, row in enumerate(reader.dicts(f"simtra/simtra_profile_{code}.csv"), 1):
+			yield "simtra_profiles", (batch_id, row_no, row["angle_deg"], row["energy_ev"], row["pt_flux"], row["au_flux"])
+
+	h = cfg["hierarchy"]
+	fa = cfg["friction_assignment"]
+	wafer_id = coupon_id = track_id = 0
+	for batch in cfg["deposition_matrix"]:
+		b_code = batch["batch_code"]
+		for w in range(1, h["wafers_per_batch"] + 1):
+			wafer_id += 1
+			wafer_dir = f"batch_{b_code}/wafer_W{w}"
+			info = reader.dicts(f"{wafer_dir}/wafer_info.csv")[0]
+			yield "wafers", (
+				wafer_id, info["wafer_code"], batch_ids[b_code], info["wafer_index"],
+				info["deposition_date"], info["coupons"], info["friction_coupons"], info["reserve_coupons"],
+			)
+			for c in range(1, h["coupons_per_wafer"] + 1):
+				coupon_id += 1
+				rel_dir = f"{wafer_dir}/coupon_C{c:02d}"
+				info = reader.dicts(f"{rel_dir}/coupon_info.csv")[0]
+				is_friction = info["run_code"] != "RESERVE"
+				yield "coupons", (
+					coupon_id, info["coupon_code"], wafer_id, batch_ids[b_code],
+					info["grid_row"], info["grid_col"], info["thickness_um"], info["ra_nm"],
+					info["au_wtpct_mean"], run_ids[info["run_code"]] if is_friction else None,
+					info.get("plate"), info.get("probe"), info.get("square"),
+				)
+				for row in reader.dicts(f"{rel_dir}/xrf_map.csv"):
+					yield "xrf_points", (coupon_id, row["grid_x"], row["grid_y"], row["pt_wtpct"], row["au_wtpct"])
+				for row in reader.dicts(f"{rel_dir}/profilometry.csv"):
+					yield "profilometry_points", (coupon_id, row["grid_x"], row["grid_y"], row["thickness_um"])
+				for row in reader.dicts(f"{rel_dir}/nanoindentation.csv"):
+					yield "nanoindentation", (
+						coupon_id, row["indent_id"], row["x_um"], row["y_um"],
+						row["hardness_gpa"], row["reduced_modulus_gpa"], row["max_load_mn"],
+					)
+				afm = reader.dicts(f"{rel_dir}/afm.csv")[0]
+				yield "afm", (coupon_id, afm["ra_nm"], afm["rq_nm"], afm["image_file"])
+				if not is_friction:
+					continue
+				for t in range(1, fa["tracks_per_friction_coupon"] + 1):
+					track_id += 1
+					track_dir = f"{rel_dir}/track_T{t}"
+					tinfo = reader.dicts(f"{track_dir}/track_info.csv")[0]
+					yield "tracks", (
+						track_id, tinfo["track_code"], coupon_id, run_ids[tinfo["run_code"]],
+						tinfo["environment"], tinfo["load_mn"], tinfo["stroke_mm"], tinfo["speed_mm_s"],
+						tinfo["counterface_id"], tinfo["started_at"],
+					)
+					cofs: list[float] = []
+					for row in reader.dicts(f"{track_dir}/cof_vs_cycle.csv"):
+						cofs.append(row["cof"])
+						yield "friction_cycles", (track_id, row["cycle"], row["cof"])
+					ss_mean, ss_std, run_in = summarize_track(cofs)
+					yield "track_summary", (track_id, ss_mean, ss_std, run_in)
+					pt = 0
+					last_cycle = None
+					for row in reader.dicts(f"{track_dir}/friction_loops.csv"):
+						pt = pt + 1 if row["cycle"] == last_cycle else 1
+						last_cycle = row["cycle"]
+						yield "friction_loop_points", (track_id, row["cycle"], pt, row["position_um"], row["friction_force_mn"])
+					wear = reader.dicts(f"{track_dir}/wear.csv")[0]
+					yield "wear", (track_id, wear["wear_volume_um3"], wear["k_archard"], wear["sliding_distance_m"])
+		logger.info("batch %s streamed (through coupon %d, track %d)", b_code, coupon_id, track_id)

common/results.py

@@ -0,0 +1,73 @@
+"""Canonical benchmark result handling (spec 08, test-pipeline-validation).
+
+Every implementation of a benchmark query returns the same canonical rows:
+tuples of str / int / float columns. Rows are compared SORTED; floats agree
+within 1e-9. The canonical expected artifacts are produced from PostgreSQL
+and cross-validated against SQLite (task 08); every measured run (task 09)
+re-validates its rows against them.
+
+Two serializations:
+- full: floats via repr() - lossless, used for stdout and expected/*.rows.csv;
+- rounded (6 decimals): used ONLY for the sha256 fingerprint, so engines
+  whose aggregate arithmetic differs in the last bits hash identically.
+The tolerant row comparison is the authoritative check; the sha256 is a
+compact fingerprint for reports and markers.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import sys
+
+from common.corpus import coerce
+
+FLOAT_TOL = 1e-9
+
+Row = tuple
+
+
+def sort_rows(rows: list[Row]) -> list[Row]:
+	return sorted(rows)
+
+
+def _fmt(value, rounded: bool) -> str:
+	if isinstance(value, float):
+		return f"{round(value, 6):.6f}" if rounded else repr(value)
+	return str(value)
+
+
+def serialize(rows: list[Row], rounded: bool = False) -> str:
+	return "\n".join(",".join(_fmt(v, rounded) for v in row) for row in rows) + "\n"
+
+
+def sha256_of(rows: list[Row]) -> str:
+	return hashlib.sha256(serialize(sort_rows(rows), rounded=True).encode("ascii")).hexdigest()
+
+
+def print_rows(rows: list[Row]) -> None:
+	sys.stdout.write(serialize(sort_rows(rows)))
+
+
+def parse_rows(text: str) -> list[Row]:
+	rows = []
+	for line in text.splitlines():
+		if line:
+			rows.append(tuple(coerce(cell) for cell in line.split(",")))
+	return rows
+
+
+def compare(expected: list[Row], actual: list[Row], tol: float = FLOAT_TOL) -> str | None:
+	"""Return None when equal within tolerance, else a first-difference message."""
+	exp, act = sort_rows(expected), sort_rows(actual)
+	if len(exp) != len(act):
+		return f"row count {len(act)} != expected {len(exp)}"
+	for i, (er, ar) in enumerate(zip(exp, act)):
+		if len(er) != len(ar):
+			return f"row {i}: arity {len(ar)} != {len(er)}"
+		for j, (ev, av) in enumerate(zip(er, ar)):
+			if isinstance(ev, float) or isinstance(av, float):
+				if abs(float(ev) - float(av)) > tol:
+					return f"row {i} col {j}: {av!r} != {ev!r} (tol {tol})"
+			elif ev != av:
+				return f"row {i} col {j}: {av!r} != {ev!r}"
+	return None

common/storage_sizes.py

@@ -8,15 +8,23 @@ touches other formats' rows.
 from __future__ import annotations
 
 import logging
+import os
+import time
 from pathlib import Path
 
 logger = logging.getLogger(__name__)
 
 HEADER = "format,variant,bytes"
+LOCK_TIMEOUT_S = 30.0  # converters 04-07 may run in parallel; the update is a
+LOCK_POLL_S = 0.2      # read-modify-write, so it is serialized via a lock file
 
 
 def update_storage_sizes(bench_dir: Path, fmt: str, variants: list[tuple[str, int]]) -> Path:
+	bench_dir.mkdir(parents=True, exist_ok=True)
 	path = bench_dir / "storage_sizes.csv"
+	lock = bench_dir / "storage_sizes.lock"
+	fd = _acquire(lock)
+	try:
 		kept: list[str] = []
 		if path.exists():
 			lines = path.read_text(encoding="ascii").splitlines()
@@ -24,8 +32,21 @@ def update_storage_sizes(bench_dir: Path, fmt: str, variants: list[tuple[str, in
 				raise ValueError(f"{path}: unexpected header {lines[0]!r}")
 			kept = [ln for ln in lines[1:] if ln and ln.split(",", 1)[0] != fmt]
 		rows = kept + [f"{fmt},{variant},{nbytes}" for variant, nbytes in variants]
-	bench_dir.mkdir(parents=True, exist_ok=True)
 		with open(path, "w", encoding="ascii", newline="\n") as fh:
 			fh.write(HEADER + "\n" + "\n".join(rows) + "\n")
+	finally:
+		os.close(fd)
+		lock.unlink()
 	logger.info("storage_sizes.csv: %s -> %s", fmt, ", ".join(f"{v}={b}" for v, b in variants))
 	return path
+
+
+def _acquire(lock: Path) -> int:
+	deadline = time.monotonic() + LOCK_TIMEOUT_S
+	while True:
+		try:
+			return os.open(lock, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
+		except FileExistsError:
+			if time.monotonic() > deadline:
+				raise TimeoutError(f"storage_sizes lock held too long: {lock}")
+			time.sleep(LOCK_POLL_S)

convert_json.py

@@ -30,6 +30,7 @@ from common.pipeline import (
 	check_dependencies,
 	load_lab_config,
 	parse_task_args,
+	process_metrics,
 	remove_stale_marker,
 	write_marker,
 )
@@ -265,7 +266,7 @@ def main() -> int:
 			json_root / "hybrid" / "dataset.jsonld",
 		])
 		update_storage_sizes(out_root / "bench", "json", [("full", conv.full_bytes), ("hybrid", hybrid_bytes)])
-		marker_path = write_marker(out_root, TASK_ID, {
+		entries = {
 			"full_files": conv.full_files,
 			"full_bytes": conv.full_bytes,
 			"hybrid_bytes": hybrid_bytes,
@@ -273,7 +274,9 @@ def main() -> int:
 			"tracks": conv.counts["tracks"],
 			"cycle_rows": conv.counts["cycle_rows"],
 			"loop_points": conv.counts["loop_rows"],
-		})
+		}
+		entries.update(process_metrics())
+		marker_path = write_marker(out_root, TASK_ID, entries)
 	except Exception:
 		logger.critical("task %s failed", TASK_ID, exc_info=True)
 		return 1

convert_postgresql.py

@@ -0,0 +1,334 @@
+"""Task 06 - Convert the CSV corpus to PostgreSQL.
+
+Loads the same logical schema as task 05 into the PostgreSQL 16 database
+named by the TRIBO_PG_DSN environment variable: COPY FROM STDIN, RANGE
+partitioning of the two bulk tables by track_id (12 partitions of 120
+tracks), constraints and indexes created after the load (variant B:
+enforced foreign keys everywhere - docs/research/Tribology_FK_Architecture.html),
+track_summary as a materialized view (algorithm fixed in
+docs/rules/db-sql-schema.md section 5, cross-checked against
+common/track_summary.py). Host CPU/RAM during the load is sampled from the
+database host's Prometheus (TRIBO_PROM_URL). Appends live (and, when
+pg_dump is available, dump) sizes to storage_sizes.csv and writes
+./out/.done/06.ok. Spec: docs/specs/06_convert_postgresql.md.
+"""
+
+from __future__ import annotations
+
+import logging
+import shutil
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+import psycopg
+
+from common.corpus import CorpusReader, ValidationError
+from common.monitoring import get_prom_url, host_window_metrics
+from common.pg import get_dsn
+from common.pipeline import (
+	check_dependencies,
+	load_lab_config,
+	parse_task_args,
+	process_metrics,
+	remove_stale_marker,
+	write_marker,
+)
+from common.relational import COLUMNS, TABLES, expected_counts, stream_rows
+from common.storage_sizes import update_storage_sizes
+from common.track_summary import summarize_track
+
+logger = logging.getLogger(__name__)
+
+TASK_ID = "06"
+DEPENDS_ON = ["01", "03"]
+
+BATCH_ROWS = 50000  # rows per COPY chunk (mirrors the 50k batching of spec 05)
+PARTITIONS = 12  # RANGE partitions of 120 tracks each, spec 06
+TRACKS_PER_PARTITION = 120
+SUMMARY_PARITY_TRACKS = (1, 720, 1440)  # matview cross-checked vs common/track_summary.py
+PARITY_TOLERANCE = 1e-9  # cross-engine float tolerance, test-pipeline-validation
+
+# Float columns are DOUBLE PRECISION, not REAL: the corpus is float64 and
+# REAL (float4) would break the 1e-9 cross-engine result parity that the
+# whole benchmark depends on (db-sql-schema section 6; spec updated).
+CREATE_TABLES = [
+	"CREATE TABLE instruments (instrument_id INTEGER NOT NULL, instrument_code TEXT NOT NULL, name TEXT NOT NULL, role TEXT NOT NULL)",
+	"CREATE TABLE batches (batch_id INTEGER NOT NULL, batch_code TEXT NOT NULL, pt_gun_tilt_deg INTEGER NOT NULL, au_gun_tilt_deg INTEGER NOT NULL, pt_power_w INTEGER NOT NULL, au_power_w INTEGER NOT NULL, pt_discharge_v INTEGER NOT NULL, au_discharge_v INTEGER NOT NULL, deposition_date DATE NOT NULL)",
+	"CREATE TABLE runs (run_id INTEGER NOT NULL, run_code TEXT NOT NULL, environment TEXT NOT NULL, date DATE NOT NULL, plates INTEGER NOT NULL, operator TEXT NOT NULL)",
+	"CREATE TABLE wafers (wafer_id INTEGER NOT NULL, wafer_code TEXT NOT NULL, batch_id INTEGER NOT NULL, wafer_index INTEGER NOT NULL, deposition_date DATE NOT NULL, coupons INTEGER NOT NULL, friction_coupons INTEGER NOT NULL, reserve_coupons INTEGER NOT NULL)",
+	"CREATE TABLE coupons (coupon_id INTEGER NOT NULL, coupon_code TEXT NOT NULL, wafer_id INTEGER NOT NULL, batch_id INTEGER NOT NULL, grid_row INTEGER NOT NULL, grid_col INTEGER NOT NULL, thickness_um DOUBLE PRECISION NOT NULL, ra_nm DOUBLE PRECISION NOT NULL, au_wtpct_mean DOUBLE PRECISION NOT NULL, run_id INTEGER, plate INTEGER, probe INTEGER, square INTEGER)",
+	"CREATE TABLE tracks (track_id INTEGER NOT NULL, track_code TEXT NOT NULL, coupon_id INTEGER NOT NULL, run_id INTEGER NOT NULL, environment TEXT NOT NULL, load_mn DOUBLE PRECISION NOT NULL, stroke_mm DOUBLE PRECISION NOT NULL, speed_mm_s DOUBLE PRECISION NOT NULL, counterface_id TEXT NOT NULL, started_at TIMESTAMPTZ NOT NULL)",
+	"CREATE TABLE simtra_profiles (batch_id INTEGER NOT NULL, row_no INTEGER NOT NULL, angle_deg DOUBLE PRECISION NOT NULL, energy_ev DOUBLE PRECISION NOT NULL, pt_flux DOUBLE PRECISION NOT NULL, au_flux DOUBLE PRECISION NOT NULL)",
+	"CREATE TABLE xrf_points (coupon_id INTEGER NOT NULL, grid_x INTEGER NOT NULL, grid_y INTEGER NOT NULL, pt_wtpct DOUBLE PRECISION NOT NULL, au_wtpct DOUBLE PRECISION NOT NULL)",
+	"CREATE TABLE profilometry_points (coupon_id INTEGER NOT NULL, grid_x INTEGER NOT NULL, grid_y INTEGER NOT NULL, thickness_um DOUBLE PRECISION NOT NULL)",
+	"CREATE TABLE nanoindentation (coupon_id INTEGER NOT NULL, indent_id INTEGER NOT NULL, x_um DOUBLE PRECISION NOT NULL, y_um DOUBLE PRECISION NOT NULL, hardness_gpa DOUBLE PRECISION NOT NULL, reduced_modulus_gpa DOUBLE PRECISION NOT NULL, max_load_mn DOUBLE PRECISION NOT NULL)",
+	"CREATE TABLE afm (coupon_id INTEGER NOT NULL, ra_nm DOUBLE PRECISION NOT NULL, rq_nm DOUBLE PRECISION NOT NULL, image_file TEXT NOT NULL)",
+	"CREATE TABLE friction_cycles (track_id INTEGER NOT NULL, cycle INTEGER NOT NULL, cof DOUBLE PRECISION NOT NULL) PARTITION BY RANGE (track_id)",
+	"CREATE TABLE friction_loop_points (track_id INTEGER NOT NULL, cycle INTEGER NOT NULL, pt INTEGER NOT NULL, position_um DOUBLE PRECISION NOT NULL, friction_force_mn DOUBLE PRECISION NOT NULL) PARTITION BY RANGE (track_id)",
+	"CREATE TABLE wear (track_id INTEGER NOT NULL, wear_volume_um3 DOUBLE PRECISION NOT NULL, k_archard DOUBLE PRECISION NOT NULL, sliding_distance_m DOUBLE PRECISION NOT NULL)",
+]
+
+CONSTRAINTS = [
+	"ALTER TABLE instruments ADD PRIMARY KEY (instrument_id)",
+	"ALTER TABLE instruments ADD UNIQUE (instrument_code)",
+	"ALTER TABLE batches ADD PRIMARY KEY (batch_id)",
+	"ALTER TABLE batches ADD UNIQUE (batch_code)",
+	"ALTER TABLE runs ADD PRIMARY KEY (run_id)",
+	"ALTER TABLE runs ADD UNIQUE (run_code)",
+	"ALTER TABLE wafers ADD PRIMARY KEY (wafer_id)",
+	"ALTER TABLE wafers ADD UNIQUE (wafer_code)",
+	"ALTER TABLE coupons ADD PRIMARY KEY (coupon_id)",
+	"ALTER TABLE coupons ADD UNIQUE (coupon_code)",
+	"ALTER TABLE tracks ADD PRIMARY KEY (track_id)",
+	"ALTER TABLE tracks ADD UNIQUE (track_code)",
+	"ALTER TABLE simtra_profiles ADD PRIMARY KEY (batch_id, row_no)",
+	"ALTER TABLE xrf_points ADD PRIMARY KEY (coupon_id, grid_x, grid_y)",
+	"ALTER TABLE profilometry_points ADD PRIMARY KEY (coupon_id, grid_x, grid_y)",
+	"ALTER TABLE nanoindentation ADD PRIMARY KEY (coupon_id, indent_id)",
+	"ALTER TABLE afm ADD PRIMARY KEY (coupon_id)",
+	"ALTER TABLE friction_cycles ADD PRIMARY KEY (track_id, cycle)",
+	"ALTER TABLE friction_loop_points ADD PRIMARY KEY (track_id, cycle, pt)",
+	"ALTER TABLE wear ADD PRIMARY KEY (track_id)",
+	"ALTER TABLE wafers ADD FOREIGN KEY (batch_id) REFERENCES batches",
+	"ALTER TABLE coupons ADD FOREIGN KEY (wafer_id) REFERENCES wafers",
+	"ALTER TABLE coupons ADD FOREIGN KEY (batch_id) REFERENCES batches",
+	"ALTER TABLE coupons ADD FOREIGN KEY (run_id) REFERENCES runs",
+	"ALTER TABLE tracks ADD FOREIGN KEY (coupon_id) REFERENCES coupons",
+	"ALTER TABLE tracks ADD FOREIGN KEY (run_id) REFERENCES runs",
+	"ALTER TABLE simtra_profiles ADD FOREIGN KEY (batch_id) REFERENCES batches",
+	"ALTER TABLE xrf_points ADD FOREIGN KEY (coupon_id) REFERENCES coupons",
+	"ALTER TABLE profilometry_points ADD FOREIGN KEY (coupon_id) REFERENCES coupons",
+	"ALTER TABLE nanoindentation ADD FOREIGN KEY (coupon_id) REFERENCES coupons",
+	"ALTER TABLE afm ADD FOREIGN KEY (coupon_id) REFERENCES coupons",
+	"ALTER TABLE friction_cycles ADD FOREIGN KEY (track_id) REFERENCES tracks",
+	"ALTER TABLE friction_loop_points ADD FOREIGN KEY (track_id) REFERENCES tracks",
+	"ALTER TABLE wear ADD FOREIGN KEY (track_id) REFERENCES tracks",
+]
+
+INDEX_DDL = [
+	"CREATE INDEX ix_coupons_au_wtpct_mean ON coupons(au_wtpct_mean)",  # Q2 selective filter
+	"CREATE INDEX ix_tracks_run_id ON tracks(run_id)",  # Q5/Q7 joins to runs
+	"CREATE INDEX ix_tracks_coupon_id ON tracks(coupon_id)",  # Q3/Q6 joins to coupons
+	"CREATE INDEX ix_runs_environment ON runs(environment)",  # Q7 grouping
+	"CREATE INDEX ix_tracks_environment_load ON tracks(environment, load_mn)",  # Q4 filter
+	# Q5 pruning within partitions; cascades to every partition (spec 06).
+	# nanoindentation(coupon_id) from the spec is omitted: the PK prefix serves it.
+	"CREATE INDEX ix_brin_friction_cycles_cycle ON friction_cycles USING BRIN (cycle)",
+]
+
+# Materialized view semantics fixed in docs/rules/db-sql-schema.md section 5.
+MATVIEW_SQL = """
+CREATE MATERIALIZED VIEW track_summary AS
+WITH tail AS (
+	SELECT track_id, avg(cof) AS m, stddev_pop(cof) AS s
+	FROM friction_cycles
+	WHERE cycle >= 501
+	GROUP BY track_id
+),
+run_in AS (
+	SELECT fc.track_id,
+	       COALESCE(MIN(fc.cycle) FILTER (WHERE abs(fc.cof - t.m) < 2 * t.s), 500) AS run_in_cycles
+	FROM friction_cycles fc
+	JOIN tail t USING (track_id)
+	GROUP BY fc.track_id
+)
+SELECT r.track_id,
+       avg(fc.cof) AS cof_ss_mean,
+       stddev_pop(fc.cof) AS cof_ss_std,
+       r.run_in_cycles
+FROM run_in r
+JOIN friction_cycles fc ON fc.track_id = r.track_id AND fc.cycle > r.run_in_cycles
+GROUP BY r.track_id, r.run_in_cycles
+"""
+
+
+def partition_ddl() -> list[str]:
+	ddl = []
+	for parent in ("friction_cycles", "friction_loop_points"):
+		for i in range(PARTITIONS):
+			lo = i * TRACKS_PER_PARTITION + 1
+			hi = lo + TRACKS_PER_PARTITION
+			ddl.append(f"CREATE TABLE {parent}_p{i + 1:02d} PARTITION OF {parent} FOR VALUES FROM ({lo}) TO ({hi})")
+	return ddl
+
+
+class PgLoader:
+	def __init__(self, conn: psycopg.Connection) -> None:
+		self.conn = conn
+		self.buffers: dict[str, list[tuple]] = {t: [] for t in TABLES}
+		self.summary_parity: dict[int, tuple] = {}
+
+	def create_schema(self) -> None:
+		with self.conn.cursor() as cur:
+			cur.execute("DROP MATERIALIZED VIEW IF EXISTS track_summary")
+			for table in reversed(TABLES):
+				if table != "track_summary":
+					cur.execute(f"DROP TABLE IF EXISTS {table} CASCADE")
+			for ddl in CREATE_TABLES + partition_ddl():
+				cur.execute(ddl)
+		self.conn.commit()
+		logger.info("schema created: %d tables, %d partitions", len(CREATE_TABLES), 2 * PARTITIONS)
+
+	def load(self, cfg: dict, reader: CorpusReader) -> None:
+		pending = 0
+		for table, row in stream_rows(cfg, reader):
+			if table == "track_summary":
+				# computed server-side as a materialized view; keep a sample
+				# of the Python values for the cross-engine parity check
+				if row[0] in SUMMARY_PARITY_TRACKS:
+					self.summary_parity[row[0]] = row
+				continue
+			self.buffers[table].append(row)
+			pending += 1
+			if pending >= BATCH_ROWS:
+				self.flush_all()
+				pending = 0
+		self.flush_all()
+
+	def flush_all(self) -> None:
+		for table in TABLES:
+			buf = self.buffers.get(table)
+			if not buf:
+				continue
+			cols = ", ".join(COLUMNS[table])
+			with self.conn.cursor() as cur:
+				with cur.copy(f"COPY {table} ({cols}) FROM STDIN") as copy:
+					for row in buf:
+						copy.write_row(row)
+			buf.clear()
+		self.conn.commit()
+
+	def finalize(self) -> None:
+		with self.conn.cursor() as cur:
+			for ddl in CONSTRAINTS:
+				cur.execute(ddl)
+			for ddl in INDEX_DDL:
+				cur.execute(ddl)
+			cur.execute(MATVIEW_SQL)
+			cur.execute("CREATE UNIQUE INDEX ix_track_summary_track_id ON track_summary(track_id)")  # Q2/Q3 joins
+		self.conn.commit()
+		logger.info("constraints, indexes and track_summary materialized view created")
+
+	def validate(self, cfg: dict) -> dict[str, int]:
+		counts: dict[str, int] = {}
+		with self.conn.cursor() as cur:
+			for table, exp in expected_counts(cfg).items():
+				got = cur.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
+				counts[table] = got
+				if got != exp:
+					raise ValidationError(f"row count mismatch: {table}: loaded {got} != expected {exp}")
+			reserve = cur.execute("SELECT COUNT(*) FROM coupons WHERE run_id IS NULL").fetchone()[0]
+			if reserve != cfg["friction_assignment"]["reserve_coupons_total"]:
+				raise ValidationError(f"reserve coupons: {reserve} != expected")
+			for track_id, py_row in self.summary_parity.items():
+				db_row = cur.execute(
+					"SELECT cof_ss_mean, cof_ss_std, run_in_cycles FROM track_summary WHERE track_id = %s",
+					(track_id,),
+				).fetchone()
+				if db_row is None:
+					raise ValidationError(f"track_summary missing track {track_id}")
+				if (abs(db_row[0] - py_row[1]) > PARITY_TOLERANCE
+						or abs(db_row[1] - py_row[2]) > PARITY_TOLERANCE
+						or db_row[2] != py_row[3]):
+					raise ValidationError(
+						f"track_summary parity failed for track {track_id}: db={db_row} py={py_row[1:]}"
+					)
+		logger.info(
+			"all %d table counts match, %d reserve coupons, matview parity ok for tracks %s",
+			len(counts), reserve, list(self.summary_parity),
+		)
+		return counts
+
+	def table_sizes(self) -> dict[str, int]:
+		sizes: dict[str, int] = {}
+		with self.conn.cursor() as cur:
+			for table in TABLES:
+				total = cur.execute(
+					"""SELECT COALESCE(pg_total_relation_size(%s::regclass), 0)
+					 + COALESCE((SELECT sum(pg_total_relation_size(inhrelid))
+					             FROM pg_inherits WHERE inhparent = %s::regclass), 0)""",
+					(table, table),
+				).fetchone()[0]
+				sizes[table] = int(total)
+		return sizes
+
+
+def make_dump(dsn: str, dump_path: Path) -> int:
+	pg_dump = shutil.which("pg_dump")
+	if not pg_dump:
+		logger.warning("pg_dump skipped (client not installed on this host; produce the dump on the database host)")
+		return 0
+	dump_path.parent.mkdir(parents=True, exist_ok=True)
+	subprocess.run([pg_dump, "--format=custom", f"--file={dump_path}", dsn], check=True)
+	return dump_path.stat().st_size
+
+
+def main() -> int:
+	args = parse_task_args("Task 06: convert the CSV corpus to PostgreSQL")
+	out_root: Path = args.out_root
+	dsn = get_dsn()
+	try:
+		check_dependencies(out_root, DEPENDS_ON)
+		remove_stale_marker(out_root, TASK_ID)
+		cfg = load_lab_config(out_root)
+
+		t_start = time.time()
+		with psycopg.connect(dsn) as conn:
+			database = conn.info.dbname
+			logger.info("connected to database %r on %s", database, conn.info.host)
+			loader = PgLoader(conn)
+			loader.create_schema()
+			loader.load(cfg, CorpusReader(out_root / "csv"))
+			load_done = time.time()
+			loader.finalize()
+			counts = loader.validate(cfg)
+			sizes = loader.table_sizes()
+		with psycopg.connect(dsn, autocommit=True) as conn:
+			conn.execute("VACUUM ANALYZE")
+		t_end = time.time()
+
+		live_bytes = sum(sizes.values())
+		variants = [("live", live_bytes)]
+		dump_bytes = make_dump(dsn, out_root / "pg" / "tribo.dump")
+		if dump_bytes:
+			variants.append(("dump", dump_bytes))
+		update_storage_sizes(out_root / "bench", "pg", variants)
+
+		entries = {
+			"database": database,
+			"tables": len(counts),
+			"total_rows": sum(counts.values()),
+			"live_bytes": live_bytes,
+			"friction_cycles_bytes": sizes["friction_cycles"],
+			"friction_loop_points_bytes": sizes["friction_loop_points"],
+			"copy_seconds": round(load_done - t_start, 1),
+			"total_seconds": round(t_end - t_start, 1),
+			"dump": dump_bytes if dump_bytes else "skipped (pg_dump unavailable on this host)",
+		}
+		entries.update(process_metrics())  # client-side converter cost
+		prom_url = get_prom_url()
+		if prom_url:
+			metrics = host_window_metrics(prom_url, t_start, t_end)
+			if metrics:
+				entries.update(metrics)
+				logger.info("db host during load: %s", metrics)
+		marker_path = write_marker(out_root, TASK_ID, entries)
+	except Exception:
+		logger.critical("task %s failed", TASK_ID, exc_info=True)
+		return 1
+
+	print(f"task 06 ok: database {database} ({live_bytes / (1024 * 1024):.1f} MiB live, "
+		f"{len(counts)} tables, {sum(counts.values())} rows, {entries['total_seconds']} s)")
+	print(f"bulk rows: cycles={counts['friction_cycles']} loop_points={counts['friction_loop_points']} "
+		f"track_summary={counts['track_summary']}")
+	if prom_url and "host_cpu_avg_pct" in entries:
+		print(f"db host: cpu avg {entries['host_cpu_avg_pct']}% max {entries['host_cpu_max_pct']}%, "
+			f"ram peak {entries['host_ram_used_peak_mb']} MB (delta {entries['host_ram_used_delta_mb']} MB)")
+	print(f"marker: {marker_path}")
+	return 0
+
+
+if __name__ == "__main__":
+	sys.exit(main())

convert_rdf.py

@@ -0,0 +1,395 @@
+"""Task 07 - Convert the CSV corpus to RDF.
+
+Streams the corpus into ./out/rdf/dataset.nt.gz (N-Triples, gzip) and
+./out/rdf/dataset.ttl (Turtle) with the task 04 vocabulary (tribo# + PROV-O
+for started_at), then bulk-loads the N-Triples into an embedded on-disk
+oxigraph store (./out/rdf/oxigraph_store/) and smoke-tests it with SPARQL
+COUNT queries. Bulk points (cycles, loop points, map points) carry no
+rdf:type - the containing predicate types them (documented decision per
+spec 07), which is why the triple count lands below the spec's 25-45M
+sketch. Appends serialization and store sizes to storage_sizes.csv and
+writes ./out/.done/07.ok. Spec: docs/specs/07_convert_rdf.md.
+"""
+
+from __future__ import annotations
+
+import gzip
+import io
+import logging
+import shutil
+import sys
+import time
+from pathlib import Path
+
+from pyoxigraph import Store
+
+try:
+	from pyoxigraph import RdfFormat
+	NT_FORMAT = RdfFormat.N_TRIPLES
+except ImportError:  # pyoxigraph < 0.5 uses MIME strings
+	NT_FORMAT = "application/n-triples"
+
+from common.corpus import CorpusReader, ValidationError
+from common.pipeline import (
+	check_dependencies,
+	load_lab_config,
+	parse_task_args,
+	process_metrics,
+	remove_stale_marker,
+	write_marker,
+)
+from common.storage_sizes import update_storage_sizes
+
+logger = logging.getLogger(__name__)
+
+TASK_ID = "07"
+DEPENDS_ON = ["01", "03"]
+
+VOCAB = "https://sandia.gov/ontology/tribology#"
+ID_BASE = "https://sandia.gov/ontology/tribology/id/"
+XSD = "http://www.w3.org/2001/XMLSchema#"
+
+PREFIXES = {
+	"tribo": VOCAB,
+	"batch": ID_BASE + "batch/",
+	"wafer": ID_BASE + "wafer/",
+	"coupon": ID_BASE + "coupon/",
+	"track": ID_BASE + "track/",
+	"run": ID_BASE + "run/",
+	"instrument": ID_BASE + "instrument/",
+	"simtra": ID_BASE + "simtra/",
+	"prov": "http://www.w3.org/ns/prov#",
+	"xsd": XSD,
+}
+
+INT_COLS = {
+	"cycle", "grid_x", "grid_y", "indent_id", "plates", "plate", "probe", "square",
+	"row_no", "wafer_index", "coupons", "friction_coupons", "reserve_coupons",
+	"pt_gun_tilt_deg", "au_gun_tilt_deg", "pt_power_w", "au_power_w",
+	"pt_discharge_v", "au_discharge_v",
+}
+DOUBLE_COLS = {
+	"cof", "position_um", "friction_force_mn", "pt_wtpct", "au_wtpct", "thickness_um",
+	"x_um", "y_um", "hardness_gpa", "reduced_modulus_gpa", "max_load_mn", "ra_nm",
+	"rq_nm", "wear_volume_um3", "k_archard", "sliding_distance_m", "load_mn",
+	"stroke_mm", "speed_mm_s", "angle_deg", "energy_ev", "pt_flux", "au_flux",
+	"au_wtpct_mean",
+}
+DATE_COLS = {"date", "deposition_date"}
+
+MEASUREMENTS = [
+	("xrf_map.csv", "xrf", "CompositionMeasurement", "m4_tornado"),
+	("profilometry.csv", "profilometry", "ProfilometryMeasurement", "profilometer"),
+	("nanoindentation.csv", "nanoindentation", "NanoindentationMeasurement", "ti980"),
+]
+
+
+def lit(col: str, token: str) -> tuple:
+	if col in INT_COLS:
+		return ("l", token, "integer")
+	if col in DOUBLE_COLS:
+		return ("l", token, "double")
+	if col in DATE_COLS:
+		return ("l", token, "date")
+	if col == "started_at":
+		return ("l", token, "dateTime")
+	return ("l", token, None)
+
+
+def iri(curie: str) -> tuple:
+	return ("i", curie)
+
+
+def node(pairs: list) -> tuple:
+	return ("n", pairs)
+
+
+class RdfEmitter:
+	"""Writes the same subject blocks to N-Triples and Turtle streams."""
+
+	def __init__(self, nt_fh, ttl_fh) -> None:
+		self.nt = nt_fh
+		self.ttl = ttl_fh
+		self.triples = 0
+		self._blank = 0
+		for prefix, base in PREFIXES.items():
+			self.ttl.write(f"@prefix {prefix}: <{base}> .\n")
+		self.ttl.write("\n")
+
+	def _full(self, curie: str) -> str:
+		prefix, local = curie.split(":", 1)
+		return f"<{PREFIXES[prefix]}{local}>"
+
+	def _pred_nt(self, pred: str) -> str:
+		if pred == "a":
+			return "<http://www.w3.org/1999/02/22-rdf-syntax-ns#type>"
+		return self._full(pred)
+
+	@staticmethod
+	def _escape(text: str) -> str:
+		return text.replace("\\", "\\\\").replace('"', '\\"')
+
+	def _obj_nt(self, obj: tuple) -> str:
+		kind = obj[0]
+		if kind == "i":
+			return self._full(obj[1])
+		if kind == "l":
+			_, token, dtype = obj
+			if dtype is None:
+				return f'"{self._escape(token)}"'
+			return f'"{token}"^^<{XSD}{dtype}>'
+		self._blank += 1
+		label = f"_:b{self._blank}"
+		for pred, sub_obj in obj[1]:
+			self.nt.write(f"{label} {self._pred_nt(pred)} {self._obj_nt(sub_obj)} .\n")
+			self.triples += 1
+		return label
+
+	def _obj_ttl(self, obj: tuple) -> str:
+		kind = obj[0]
+		if kind == "i":
+			return obj[1]
+		if kind == "l":
+			_, token, dtype = obj
+			if dtype is None:
+				return f'"{self._escape(token)}"'
+			if dtype == "integer":
+				return token
+			return f'"{token}"^^xsd:{dtype}'
+		inner = " ; ".join(f"{p} {self._obj_ttl(o)}" for p, o in obj[1])
+		return f"[ {inner} ]"
+
+	def emit(self, subject: str, pairs: list) -> None:
+		subj_nt = self._full(subject)
+		ttl_parts = []
+		for pred, obj in pairs:
+			# N-Triples: nested nodes expand to labelled blanks first, so the
+			# referencing triple is written after the node's own triples.
+			obj_nt = self._obj_nt(obj)
+			self.nt.write(f"{subj_nt} {self._pred_nt(pred)} {obj_nt} .\n")
+			self.triples += 1
+			ttl_parts.append(f"{pred} {self._obj_ttl(obj)}")
+		self.ttl.write(subject + " " + " ;\n  ".join(ttl_parts) + " .\n")
+
+
+class RdfConverter:
+	def __init__(self, cfg: dict, csv_root: Path, emitter: RdfEmitter) -> None:
+		self.cfg = cfg
+		self.reader = CorpusReader(csv_root)
+		self.em = emitter
+		self.counts = {"coupons": 0, "tracks": 0, "cycle_nodes": 0, "loop_nodes": 0, "xrf_nodes": 0}
+
+	def raw_rows(self, relpath: str) -> list[dict[str, str]]:
+		header, rows = self.reader.read(relpath)
+		keys = [h.lower() for h in header]
+		return [dict(zip(keys, row)) for row in rows]
+
+	def prop_pairs(self, row: dict[str, str], skip: tuple = ()) -> list:
+		return [
+			("prov:startedAtTime" if col == "started_at" else f"tribo:{col}", lit(col, token))
+			for col, token in row.items()
+			if col not in skip and token != ""
+		]
+
+	def convert(self) -> None:
+		for inst in self.cfg["instruments"]:
+			self.em.emit(f"instrument:{inst['instrument_id']}", [
+				("a", iri("tribo:Instrument")),
+				("tribo:name", ("l", inst["name"], None)),
+				("tribo:role", ("l", inst["role"], None)),
+			])
+		for row in self.raw_rows("batches.csv"):
+			code = row["batch_code"]
+			self.em.emit(f"batch:{code}", [("a", iri("tribo:Batch"))] + self.prop_pairs(row))
+			points = [
+				("tribo:hasPoint", node([("tribo:row_no", ("l", str(no), "integer"))] + self.prop_pairs(p)))
+				for no, p in enumerate(self.raw_rows(f"simtra/simtra_profile_{code}.csv"), 1)
+			]
+			self.em.emit(f"simtra:{code}", [
+				("a", iri("tribo:SimtraProfile")),
+				("tribo:performedOn", iri(f"batch:{code}")),
+				("tribo:performedBy", iri("instrument:simtra")),
+			] + points)
+		for row in self.raw_rows("runs.csv"):
+			self.em.emit(f"run:{row['run_code']}", [("a", iri("tribo:Run"))] + self.prop_pairs(row))
+
+		h = self.cfg["hierarchy"]
+		for batch in self.cfg["deposition_matrix"]:
+			b_code = batch["batch_code"]
+			for w in range(1, h["wafers_per_batch"] + 1):
+				wafer_dir = f"batch_{b_code}/wafer_W{w}"
+				info = self.raw_rows(f"{wafer_dir}/wafer_info.csv")[0]
+				self.em.emit(f"wafer:{info['wafer_code']}", [
+					("a", iri("tribo:Wafer")),
+					("tribo:partOf", iri(f"batch:{b_code}")),
+				] + self.prop_pairs(info, skip=("batch_code",)))
+				for c in range(1, h["coupons_per_wafer"] + 1):
+					self.convert_coupon(f"{wafer_dir}/coupon_C{c:02d}")
+			logger.info("batch %s converted (%d triples so far)", b_code, self.em.triples)
+
+	def convert_coupon(self, rel_dir: str) -> None:
+		info = self.raw_rows(f"{rel_dir}/coupon_info.csv")[0]
+		code = info["coupon_code"]
+		subject = f"coupon:{code}"
+		is_friction = info["run_code"] != "RESERVE"
+		pairs = [("a", iri("tribo:Coupon")), ("tribo:cutFrom", iri(f"wafer:{info['wafer_code']}"))]
+		pairs += self.prop_pairs(info, skip=("batch_code", "wafer_code", "run_code"))
+		pairs.append(("tribo:run_code", ("l", info["run_code"], None)))
+		if is_friction:
+			pairs.append(("tribo:duringRun", iri(f"run:{info['run_code']}")))
+
+		for csv_name, key, cls, inst in MEASUREMENTS:
+			points = [
+				("tribo:hasPoint" if key != "nanoindentation" else "tribo:hasIndent", node(self.prop_pairs(p)))
+				for p in self.raw_rows(f"{rel_dir}/{csv_name}")
+			]
+			if key == "xrf":
+				self.counts["xrf_nodes"] += len(points)
+			pairs.append((f"tribo:{key}", node([
+				("a", iri(f"tribo:{cls}")),
+				("tribo:performedOn", iri(subject)),
+				("tribo:performedBy", iri(f"instrument:{inst}")),
+			] + points)))
+		afm = self.raw_rows(f"{rel_dir}/afm.csv")[0]
+		pairs.append(("tribo:afm", node([
+			("a", iri("tribo:AFMMeasurement")),
+			("tribo:performedOn", iri(subject)),
+			("tribo:performedBy", iri("instrument:afm")),
+		] + self.prop_pairs(afm))))
+		self.em.emit(subject, pairs)
+		self.counts["coupons"] += 1
+
+		if is_friction:
+			for t in range(1, self.cfg["friction_assignment"]["tracks_per_friction_coupon"] + 1):
+				self.convert_track(f"{rel_dir}/track_T{t}", subject)
+
+	def convert_track(self, rel_dir: str, coupon_subject: str) -> None:
+		info = self.raw_rows(f"{rel_dir}/track_info.csv")[0]
+		pairs = [
+			("a", iri("tribo:FrictionTest")),
+			("tribo:partOf", iri(coupon_subject)),
+			("tribo:duringRun", iri(f"run:{info['run_code']}")),
+			("tribo:performedBy", iri("instrument:rapid")),
+		] + self.prop_pairs(info, skip=("track_code", "run_code"))
+		pairs.insert(1, ("tribo:track_code", ("l", info["track_code"], None)))
+
+		cycles = self.raw_rows(f"{rel_dir}/cof_vs_cycle.csv")
+		pairs += [("tribo:hasCycle", node(self.prop_pairs(row))) for row in cycles]
+		self.counts["cycle_nodes"] += len(cycles)
+		loops = self.raw_rows(f"{rel_dir}/friction_loops.csv")
+		pairs += [("tribo:hasLoopPoint", node(self.prop_pairs(row))) for row in loops]
+		self.counts["loop_nodes"] += len(loops)
+		wear = self.raw_rows(f"{rel_dir}/wear.csv")[0]
+		pairs.append(("tribo:wear", node([("a", iri("tribo:WearMeasurement"))] + self.prop_pairs(wear))))
+		self.em.emit(f"track:{info['track_code']}", pairs)
+		self.counts["tracks"] += 1
+
+	def validate_counts(self) -> None:
+		v = self.cfg["volumes"]
+		expected = {
+			"coupons": v["coupons_total"],
+			"tracks": v["tracks_total"],
+			"cycle_nodes": v["cycle_rows_total"],
+			"loop_nodes": v["loop_points_total"],
+			"xrf_nodes": v["xrf_points_total"],
+		}
+		for key, exp in expected.items():
+			if self.counts[key] != exp:
+				raise ValidationError(f"count mismatch: {key}: converted {self.counts[key]} != expected {exp}")
+		logger.info("all converted counts match lab_config volumes: %s", self.counts)
+
+
+def bulk_load(store: Store, nt_gz: Path) -> None:
+	with gzip.open(nt_gz, "rb") as fh:
+		try:
+			store.bulk_load(fh, NT_FORMAT)
+		except TypeError:
+			store.bulk_load(input=fh, format=NT_FORMAT)
+
+
+def sparql_count(store: Store, query: str) -> int:
+	solutions = store.query(query)
+	return int(next(iter(solutions))[0].value)
+
+
+def main() -> int:
+	args = parse_task_args("Task 07: convert the CSV corpus to RDF")
+	out_root: Path = args.out_root
+	rdf_root = out_root / "rdf"
+	nt_gz = rdf_root / "dataset.nt.gz"
+	ttl_path = rdf_root / "dataset.ttl"
+	store_dir = rdf_root / "oxigraph_store"
+	try:
+		check_dependencies(out_root, DEPENDS_ON)
+		remove_stale_marker(out_root, TASK_ID)
+		if rdf_root.exists():
+			logger.info("re-run: removing previous output %s", rdf_root)
+			shutil.rmtree(rdf_root)
+		rdf_root.mkdir(parents=True)
+		cfg = load_lab_config(out_root)
+
+		# mtime=0 keeps the gzip byte-identical across re-runs (data-determinism)
+		with open(nt_gz, "wb") as raw:
+			gz = gzip.GzipFile(filename="", mode="wb", fileobj=raw, compresslevel=6, mtime=0)
+			with io.TextIOWrapper(gz, encoding="ascii", newline="\n") as nt_fh, \
+					open(ttl_path, "w", encoding="ascii", newline="\n") as ttl_fh:
+				emitter = RdfEmitter(nt_fh, ttl_fh)
+				conv = RdfConverter(cfg, out_root / "csv", emitter)
+				conv.convert()
+				conv.validate_counts()
+				triples = emitter.triples
+		nt_bytes = nt_gz.stat().st_size
+		ttl_bytes = ttl_path.stat().st_size
+		logger.info("serialized %d triples: nt.gz %.1f MiB, ttl %.1f MiB",
+			triples, nt_bytes / 1048576, ttl_bytes / 1048576)
+
+		t0 = time.time()
+		store = Store(str(store_dir))
+		bulk_load(store, nt_gz)
+		store.flush()
+		load_seconds = round(time.time() - t0, 1)
+
+		total = sparql_count(store, "SELECT (COUNT(*) AS ?n) WHERE { ?s ?p ?o }")
+		if total != triples:
+			raise ValidationError(f"store holds {total} triples, serialized {triples}")
+		q1 = sparql_count(
+			store,
+			f"SELECT (COUNT(?c) AS ?n) WHERE {{ <{ID_BASE}track/B722-W2-C13-T2> <{VOCAB}hasCycle> ?c }}",
+		)
+		if q1 != cfg["volumes"]["cycles_per_track"]:
+			raise ValidationError(f"SPARQL Q1 smoke test returned {q1} cycles")
+		logger.info("store smoke tests ok: %d triples, Q1 track has %d cycles (load %.1f s)", total, q1, load_seconds)
+		del store
+		store_bytes = sum(f.stat().st_size for f in store_dir.rglob("*") if f.is_file())
+
+		update_storage_sizes(out_root / "bench", "rdf", [
+			("nt_gz", nt_bytes), ("ttl", ttl_bytes), ("store", store_bytes),
+		])
+		entries = {
+			"triples": triples,
+			"nt_gz_bytes": nt_bytes,
+			"ttl_bytes": ttl_bytes,
+			"store_bytes": store_bytes,
+			"store_load_seconds": load_seconds,
+			"coupons": conv.counts["coupons"],
+			"tracks": conv.counts["tracks"],
+			"cycle_nodes": conv.counts["cycle_nodes"],
+			"loop_nodes": conv.counts["loop_nodes"],
+		}
+		entries.update(process_metrics())
+		marker_path = write_marker(out_root, TASK_ID, entries)
+	except Exception:
+		logger.critical("task %s failed", TASK_ID, exc_info=True)
+		return 1
+
+	print(f"task 07 ok: {triples} triples; nt.gz {nt_bytes / 1048576:.1f} MiB, "
+		f"ttl {ttl_bytes / 1048576:.1f} MiB, store {store_bytes / 1048576:.1f} MiB "
+		f"(bulk load {load_seconds} s)")
+	print(f"entities: coupons={conv.counts['coupons']} tracks={conv.counts['tracks']} "
+		f"cycles={conv.counts['cycle_nodes']} loop_points={conv.counts['loop_nodes']}")
+	print(f"marker: {marker_path}")
+	return 0
+
+
+if __name__ == "__main__":
+	sys.exit(main())

convert_sqlite.py

@@ -20,11 +20,12 @@ from common.pipeline import (
 	check_dependencies,
 	load_lab_config,
 	parse_task_args,
+	process_metrics,
 	remove_stale_marker,
 	write_marker,
 )
+from common.relational import COLUMNS, TABLES, expected_counts, stream_rows
 from common.storage_sizes import update_storage_sizes
-from common.track_summary import summarize_track
 
 logger = logging.getLogger(__name__)
 
@@ -178,158 +179,38 @@ INDEX_DDL = [
 	"CREATE INDEX ix_tracks_environment_load ON tracks(environment, load_mn)",  # Q4 filter
 ]
 
-INSERTS = {
-	"instruments": "INSERT INTO instruments VALUES (?,?,?,?)",
-	"batches": "INSERT INTO batches VALUES (?,?,?,?,?,?,?,?,?)",
-	"runs": "INSERT INTO runs VALUES (?,?,?,?,?,?)",
-	"wafers": "INSERT INTO wafers VALUES (?,?,?,?,?,?,?,?)",
-	"coupons": "INSERT INTO coupons VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)",
-	"tracks": "INSERT INTO tracks VALUES (?,?,?,?,?,?,?,?,?,?)",
-	"simtra_profiles": "INSERT INTO simtra_profiles VALUES (?,?,?,?,?,?)",
-	"xrf_points": "INSERT INTO xrf_points VALUES (?,?,?,?,?)",
-	"profilometry_points": "INSERT INTO profilometry_points VALUES (?,?,?,?)",
-	"nanoindentation": "INSERT INTO nanoindentation VALUES (?,?,?,?,?,?,?)",
-	"afm": "INSERT INTO afm VALUES (?,?,?,?)",
-	"friction_cycles": "INSERT INTO friction_cycles VALUES (?,?,?)",
-	"friction_loop_points": "INSERT INTO friction_loop_points VALUES (?,?,?,?,?)",
-	"wear": "INSERT INTO wear VALUES (?,?,?,?)",
-	"track_summary": "INSERT INTO track_summary VALUES (?,?,?,?)",
-}
-
 
 class SqliteLoader:
-	def __init__(self, cfg: dict, csv_root: Path, db_path: Path) -> None:
-		self.cfg = cfg
-		self.reader = CorpusReader(csv_root)
+	def __init__(self, db_path: Path) -> None:
 		db_path.parent.mkdir(parents=True, exist_ok=True)
 		self.conn = sqlite3.connect(db_path)
 		for pragma in LOAD_PRAGMAS:
 			self.conn.execute(pragma)
 		for ddl in DDL:
 			self.conn.execute(ddl)
-		self.buffers: dict[str, list[tuple]] = {}
-		self.batch_ids: dict[str, int] = {}
-		self.run_ids: dict[str, int] = {}
+		self.inserts = {
+			t: f"INSERT INTO {t} VALUES ({','.join('?' * len(COLUMNS[t]))})" for t in TABLES
+		}
+		self.buffers: dict[str, list[tuple]] = {t: [] for t in TABLES}
 
-	def put(self, table: str, row: tuple) -> None:
-		buf = self.buffers.setdefault(table, [])
-		buf.append(row)
-		if len(buf) >= BATCH_ROWS:
-			self.flush(table)
-
-	def flush(self, table: str) -> None:
-		buf = self.buffers.get(table)
-		if buf:
-			self.conn.executemany(INSERTS[table], buf)
-			self.conn.commit()
-			buf.clear()
+	def load(self, cfg: dict, reader: CorpusReader) -> None:
+		pending = 0
+		for table, row in stream_rows(cfg, reader):
+			self.buffers[table].append(row)
+			pending += 1
+			if pending >= BATCH_ROWS:
+				self.flush_all()
+				pending = 0
+		self.flush_all()
 
 	def flush_all(self) -> None:
-		for table in list(self.buffers):
-			self.flush(table)
-
-	# --- dimension + bulk loading, in FK dependency order ---
-
-	def load_flat(self) -> None:
-		for i, inst in enumerate(self.cfg["instruments"], 1):
-			self.put("instruments", (i, inst["instrument_id"], inst["name"], inst["role"]))
-		for i, row in enumerate(self.reader.dicts("batches.csv"), 1):
-			self.batch_ids[row["batch_code"]] = i
-			self.put("batches", (
-				i, row["batch_code"], row["pt_gun_tilt_deg"], row["au_gun_tilt_deg"],
-				row["pt_power_w"], row["au_power_w"], row["pt_discharge_v"],
-				row["au_discharge_v"], row["deposition_date"],
-			))
-		for i, row in enumerate(self.reader.dicts("runs.csv"), 1):
-			self.run_ids[row["run_code"]] = i
-			self.put("runs", (i, row["run_code"], row["environment"], row["date"], row["plates"], row["operator"]))
-		self.flush_all()
-		for code, batch_id in self.batch_ids.items():
-			for row_no, row in enumerate(self.reader.dicts(f"simtra/simtra_profile_{code}.csv"), 1):
-				self.put("simtra_profiles", (batch_id, row_no, row["angle_deg"], row["energy_ev"], row["pt_flux"], row["au_flux"]))
-
-	def load_hierarchy(self) -> None:
-		h = self.cfg["hierarchy"]
-		fa = self.cfg["friction_assignment"]
-		wafer_id = 0
-		coupon_id = 0
-		track_id = 0
-		for batch in self.cfg["deposition_matrix"]:
-			b_code = batch["batch_code"]
-			batch_id = self.batch_ids[b_code]
-			for w in range(1, h["wafers_per_batch"] + 1):
-				wafer_id += 1
-				wafer_dir = f"batch_{b_code}/wafer_W{w}"
-				info = self.reader.dicts(f"{wafer_dir}/wafer_info.csv")[0]
-				self.put("wafers", (
-					wafer_id, info["wafer_code"], batch_id, info["wafer_index"],
-					info["deposition_date"], info["coupons"], info["friction_coupons"], info["reserve_coupons"],
-				))
-				self.flush("wafers")
-				for c in range(1, h["coupons_per_wafer"] + 1):
-					coupon_id += 1
-					rel_dir = f"{wafer_dir}/coupon_C{c:02d}"
-					track_id = self.load_coupon(rel_dir, coupon_id, wafer_id, batch_id, track_id, fa)
-			logger.info("batch %s loaded (through coupon %d, track %d)", b_code, coupon_id, track_id)
-		self.flush_all()
-
-	def load_coupon(self, rel_dir: str, coupon_id: int, wafer_id: int, batch_id: int, track_id: int, fa: dict) -> int:
-		info = self.reader.dicts(f"{rel_dir}/coupon_info.csv")[0]
-		is_friction = info["run_code"] != "RESERVE"
-		run_id = self.run_ids[info["run_code"]] if is_friction else None
-		self.put("coupons", (
-			coupon_id, info["coupon_code"], wafer_id, batch_id,
-			info["grid_row"], info["grid_col"], info["thickness_um"], info["ra_nm"],
-			info["au_wtpct_mean"], run_id,
-			info.get("plate"), info.get("probe"), info.get("square"),
-		))
-		self.flush("coupons")
-
-		for row in self.reader.dicts(f"{rel_dir}/xrf_map.csv"):
-			self.put("xrf_points", (coupon_id, row["grid_x"], row["grid_y"], row["pt_wtpct"], row["au_wtpct"]))
-		for row in self.reader.dicts(f"{rel_dir}/profilometry.csv"):
-			self.put("profilometry_points", (coupon_id, row["grid_x"], row["grid_y"], row["thickness_um"]))
-		for row in self.reader.dicts(f"{rel_dir}/nanoindentation.csv"):
-			self.put("nanoindentation", (
-				coupon_id, row["indent_id"], row["x_um"], row["y_um"],
-				row["hardness_gpa"], row["reduced_modulus_gpa"], row["max_load_mn"],
-			))
-		afm = self.reader.dicts(f"{rel_dir}/afm.csv")[0]
-		self.put("afm", (coupon_id, afm["ra_nm"], afm["rq_nm"], afm["image_file"]))
-
-		if is_friction:
-			for t in range(1, fa["tracks_per_friction_coupon"] + 1):
-				track_id += 1
-				self.load_track(f"{rel_dir}/track_T{t}", track_id, coupon_id)
-		return track_id
-
-	def load_track(self, rel_dir: str, track_id: int, coupon_id: int) -> None:
-		info = self.reader.dicts(f"{rel_dir}/track_info.csv")[0]
-		self.put("tracks", (
-			track_id, info["track_code"], coupon_id, self.run_ids[info["run_code"]],
-			info["environment"], info["load_mn"], info["stroke_mm"], info["speed_mm_s"],
-			info["counterface_id"], info["started_at"],
-		))
-		self.flush("tracks")
-
-		cofs: list[float] = []
-		for row in self.reader.dicts(f"{rel_dir}/cof_vs_cycle.csv"):
-			cofs.append(row["cof"])
-			self.put("friction_cycles", (track_id, row["cycle"], row["cof"]))
-		ss_mean, ss_std, run_in = summarize_track(cofs)
-		self.put("track_summary", (track_id, ss_mean, ss_std, run_in))
-
-		pt = 0
-		last_cycle = None
-		for row in self.reader.dicts(f"{rel_dir}/friction_loops.csv"):
-			pt = pt + 1 if row["cycle"] == last_cycle else 1
-			last_cycle = row["cycle"]
-			self.put("friction_loop_points", (track_id, row["cycle"], pt, row["position_um"], row["friction_force_mn"]))
-
-		wear = self.reader.dicts(f"{rel_dir}/wear.csv")[0]
-		self.put("wear", (track_id, wear["wear_volume_um3"], wear["k_archard"], wear["sliding_distance_m"]))
-
-	# --- post-load ---
+		# TABLES is FK-dependency ordered: parents flush before children.
+		for table in TABLES:
+			buf = self.buffers[table]
+			if buf:
+				self.conn.executemany(self.inserts[table], buf)
+				buf.clear()
+		self.conn.commit()
 
 	def finalize(self) -> None:
 		for ddl in INDEX_DDL:
@@ -340,39 +221,20 @@ class SqliteLoader:
 		self.conn.execute("VACUUM")
 		logger.info("indexes created, WAL enabled, ANALYZE + VACUUM done")
 
-	def validate(self) -> dict[str, int]:
-		cfg = self.cfg
-		v = cfg["volumes"]
-		expected = {
-			"instruments": len(cfg["instruments"]),
-			"batches": cfg["hierarchy"]["batches"],
-			"runs": cfg["friction_assignment"]["runs"],
-			"wafers": cfg["hierarchy"]["batches"] * cfg["hierarchy"]["wafers_per_batch"],
-			"coupons": v["coupons_total"],
-			"tracks": v["tracks_total"],
-			"simtra_profiles": cfg["hierarchy"]["batches"] * v["simtra_rows_per_batch"],
-			"xrf_points": v["xrf_points_total"],
-			"profilometry_points": v["coupons_total"] * v["profilometry_points_per_coupon"],
-			"nanoindentation": v["coupons_total"] * v["nanoindentation_indents_per_coupon"],
-			"afm": v["coupons_total"],
-			"friction_cycles": v["cycle_rows_total"],
-			"friction_loop_points": v["loop_points_total"],
-			"wear": v["tracks_total"],
-			"track_summary": v["tracks_total"],
-		}
+	def validate(self, cfg: dict) -> dict[str, int]:
 		counts: dict[str, int] = {}
-		for table, exp in expected.items():
+		for table, exp in expected_counts(cfg).items():
 			got = self.conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
 			counts[table] = got
 			if got != exp:
 				raise ValidationError(f"row count mismatch: {table}: loaded {got} != expected {exp}")
 		reserve = self.conn.execute("SELECT COUNT(*) FROM coupons WHERE run_id IS NULL").fetchone()[0]
-		if reserve != self.cfg["friction_assignment"]["reserve_coupons_total"]:
+		if reserve != cfg["friction_assignment"]["reserve_coupons_total"]:
 			raise ValidationError(f"reserve coupons: {reserve} != expected")
 		fk_violations = self.conn.execute("PRAGMA foreign_key_check").fetchall()
 		if fk_violations:
 			raise ValidationError(f"foreign_key_check reported {len(fk_violations)} violations")
-		logger.info("all %d table counts match, %d reserve coupons, foreign_key_check clean", len(expected), reserve)
+		logger.info("all %d table counts match, %d reserve coupons, foreign_key_check clean", len(counts), reserve)
 		return counts
 
 
@@ -389,16 +251,15 @@ def main() -> int:
 				stale.unlink()
 		cfg = load_lab_config(out_root)
 
-		loader = SqliteLoader(cfg, out_root / "csv", db_path)
-		loader.load_flat()
-		loader.load_hierarchy()
+		loader = SqliteLoader(db_path)
+		loader.load(cfg, CorpusReader(out_root / "csv"))
 		loader.finalize()
-		counts = loader.validate()
+		counts = loader.validate(cfg)
 		loader.conn.close()
 
 		db_bytes = db_path.stat().st_size
 		update_storage_sizes(out_root / "bench", "sqlite", [("db", db_bytes)])
-		marker_path = write_marker(out_root, TASK_ID, {
+		entries = {
 			"db_file": db_path.as_posix(),
 			"db_bytes": db_bytes,
 			"tables": len(counts),
@@ -407,7 +268,9 @@ def main() -> int:
 			"friction_loop_points": counts["friction_loop_points"],
 			"tracks": counts["tracks"],
 			"track_summary": counts["track_summary"],
-		})
+		}
+		entries.update(process_metrics())
+		marker_path = write_marker(out_root, TASK_ID, entries)
 	except Exception:
 		logger.critical("task %s failed", TASK_ID, exc_info=True)
 		return 1

docs/rules/code-python-style.md

@@ -35,7 +35,7 @@ The stdlib is the default. The ONLY permitted third-party packages:
 | `oxigraph` | 07, 09 | embedded queryable triplestore |
 | `ijson` | 08, 09 | streaming JSON parsing |
 | `psycopg` (v3) | 06, 08, 09 | PostgreSQL COPY and queries |
-| `psutil` | 09 | RSS / CPU / IO measurement |
+| `psutil` | 04-07, 09 | RSS / CPU / IO measurement; converters record their own peak RSS / CPU time into their `.done` markers for the task 10 sizing model |
 | `matplotlib` | 11 | report charts |
 | `pandas` | 08 (optional) | ONLY as the separately-measured second CSV query variant; never in the pipeline itself |
 

make_queries.py

@@ -0,0 +1,289 @@
+"""Task 08 - Benchmark query definitions (Q1-Q7 x 5 formats).
+
+Writes the runnable query files under ./out/bench/queries/ (SQL for
+SQLite/PostgreSQL, SPARQL text + python drivers for RDF, python drivers
+for CSV/JSON-LD), then produces the canonical expected results: executed
+on PostgreSQL, cross-validated against SQLite row-by-row within 1e-9, and
+stored as ./out/bench/expected/q<N>.rows.csv + q<N>.sha256. CSV, JSON and
+RDF implementations are spot-checked on Q1. Writes ./out/.done/08.ok.
+Spec: docs/specs/08_benchmark_queries.md.
+"""
+
+from __future__ import annotations
+
+import logging
+import sqlite3
+import sys
+from pathlib import Path
+
+import psycopg
+
+from common.pg import get_dsn
+from common.pipeline import (
+	check_dependencies,
+	parse_task_args,
+	process_metrics,
+	remove_stale_marker,
+	write_marker,
+)
+from common.queries import Q1_TRACK_CODE, csv_queries, json_queries, rdf_queries
+from common.results import compare, serialize, sha256_of, sort_rows
+
+logger = logging.getLogger(__name__)
+
+TASK_ID = "08"
+DEPENDS_ON = ["03", "04", "05", "06", "07"]
+
+EXACT_ROW_COUNTS = {1: 1000, 3: 480, 5: 4, 6: 480, 7: 2}  # fixed by corpus volumes
+
+# --- SQL implementations -------------------------------------------------
+# Q2/Q3/Q4/Q7 use the precomputed track_summary (the relational strength);
+# Q5 must scan raw friction_cycles in every format (spec 08).
+
+Q1_SQL = f"""-- Q1: COF vs cycle curve for track {Q1_TRACK_CODE} (point read)
+SELECT fc.cycle, fc.cof
+FROM friction_cycles fc
+JOIN tracks t ON t.track_id = fc.track_id
+WHERE t.track_code = '{Q1_TRACK_CODE}'
+ORDER BY fc.cycle;
+"""
+
+Q2_SQL = """-- Q2: steady-state COF per coupon with mean Au = 10 +/- 0.5 wt%
+SELECT c.coupon_code, t.environment, AVG(s.cof_ss_mean) AS cof_ss
+FROM coupons c
+JOIN tracks t ON t.coupon_id = c.coupon_id
+JOIN track_summary s ON s.track_id = t.track_id
+WHERE c.au_wtpct_mean BETWEEN 9.5 AND 10.5
+GROUP BY c.coupon_code, t.environment
+ORDER BY c.coupon_code;
+"""
+
+Q3_SQL = """-- Q3: hardness vs steady-state COF per friction coupon, across all batches
+SELECT c.coupon_code,
+       b.batch_code,
+       (SELECT AVG(n.hardness_gpa) FROM nanoindentation n WHERE n.coupon_id = c.coupon_id) AS hardness_gpa,
+       AVG(s.cof_ss_mean) AS cof_ss
+FROM coupons c
+JOIN batches b ON b.batch_id = c.batch_id
+JOIN tracks t ON t.coupon_id = c.coupon_id
+JOIN track_summary s ON s.track_id = t.track_id
+GROUP BY c.coupon_id, c.coupon_code, b.batch_code
+ORDER BY c.coupon_code;
+"""
+
+Q4_SQL = """-- Q4: anomaly filter - Dry N2 tracks at 100 mN with cof_ss > 0.20
+SELECT t.track_code, s.cof_ss_mean
+FROM tracks t
+JOIN track_summary s ON s.track_id = t.track_id
+WHERE t.environment = 'dry_n2' AND t.load_mn = 100 AND s.cof_ss_mean > 0.20
+ORDER BY t.track_code;
+"""
+
+# Q5: forced full scan over raw cycles; the run-in algorithm is inlined
+# (docs/rules/db-sql-schema.md section 5). SQLite has no stddev built-in,
+# so the 2-sigma band compares squares: (cof-m)^2 < 4*var  <=>  |cof-m| < 2s.
+Q5_SQLITE = """-- Q5: mean run-in cycles per batch over ALL raw cycle rows (track_summary forbidden)
+WITH tail AS (
+	SELECT track_id, AVG(cof) AS m, AVG(cof * cof) - AVG(cof) * AVG(cof) AS var
+	FROM friction_cycles
+	WHERE cycle >= 501
+	GROUP BY track_id
+),
+run_in AS (
+	SELECT fc.track_id,
+	       COALESCE(MIN(CASE WHEN (fc.cof - tl.m) * (fc.cof - tl.m) < 4 * tl.var THEN fc.cycle END), 500) AS run_in
+	FROM friction_cycles fc
+	JOIN tail tl ON tl.track_id = fc.track_id
+	GROUP BY fc.track_id
+)
+SELECT b.batch_code, AVG(r.run_in) AS avg_run_in_cycles
+FROM run_in r
+JOIN tracks t ON t.track_id = r.track_id
+JOIN coupons c ON c.coupon_id = t.coupon_id
+JOIN batches b ON b.batch_id = c.batch_id
+GROUP BY b.batch_code
+ORDER BY b.batch_code;
+"""
+
+Q5_PG = """-- Q5: mean run-in cycles per batch over ALL raw cycle rows (track_summary forbidden)
+WITH tail AS (
+	SELECT track_id, avg(cof) AS m, stddev_pop(cof) AS s
+	FROM friction_cycles
+	WHERE cycle >= 501
+	GROUP BY track_id
+),
+run_in AS (
+	SELECT fc.track_id,
+	       COALESCE(MIN(fc.cycle) FILTER (WHERE abs(fc.cof - tl.m) < 2 * tl.s), 500) AS run_in
+	FROM friction_cycles fc
+	JOIN tail tl ON tl.track_id = fc.track_id
+	GROUP BY fc.track_id
+)
+SELECT b.batch_code, AVG(r.run_in)::double precision AS avg_run_in_cycles
+FROM run_in r
+JOIN tracks t ON t.track_id = r.track_id
+JOIN coupons c ON c.coupon_id = t.coupon_id
+JOIN batches b ON b.batch_id = c.batch_id
+GROUP BY b.batch_code
+ORDER BY b.batch_code;
+"""
+
+Q6_SQL = """-- Q6: mean wear volume vs load per friction coupon
+SELECT c.coupon_code, t.load_mn, AVG(w.wear_volume_um3) AS wear_volume_um3
+FROM wear w
+JOIN tracks t ON t.track_id = w.track_id
+JOIN coupons c ON c.coupon_id = t.coupon_id
+GROUP BY c.coupon_id, c.coupon_code, t.load_mn
+ORDER BY c.coupon_code;
+"""
+
+Q7_SQL = """-- Q7: Stribeck-style mean COF by (environment, speed*load bucket) past run-in
+SELECT t.environment,
+       CAST(ROUND(t.speed_mm_s * t.load_mn) AS INTEGER) AS speed_load_bucket,
+       AVG(fc.cof) AS mean_cof
+FROM friction_cycles fc
+JOIN tracks t ON t.track_id = fc.track_id
+JOIN track_summary s ON s.track_id = fc.track_id
+WHERE fc.cycle > s.run_in_cycles
+GROUP BY t.environment, speed_load_bucket
+ORDER BY t.environment, speed_load_bucket;
+"""
+
+SQLITE_SQL = {1: Q1_SQL, 2: Q2_SQL, 3: Q3_SQL, 4: Q4_SQL, 5: Q5_SQLITE, 6: Q6_SQL, 7: Q7_SQL}
+PG_SQL = {1: Q1_SQL, 2: Q2_SQL, 3: Q3_SQL, 4: Q4_SQL, 5: Q5_PG, 6: Q6_SQL, 7: Q7_SQL}
+
+RDF_SPARQL_FILES = {
+	1: ("q1",), 2: ("q2",), 3: ("q3_hardness", "q3_cycles"),
+	4: ("q4",), 5: ("q5",), 6: ("q6",), 7: ("q7",),
+}
+
+WRAPPER = '''"""Benchmark query q{n} for the {fmt} format (generated by task 08).
+
+Prints the canonical result rows to stdout at full float precision; the
+benchmark runner (task 09) captures and validates them against
+out/bench/expected/q{n}.rows.csv.
+"""
+import sys
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[4]
+sys.path.insert(0, str(REPO_ROOT))
+
+from common.queries.{module} import q{n}
+from common.results import print_rows
+
+if __name__ == "__main__":
+	print_rows(q{n}(REPO_ROOT / "out" / {subpath}))
+'''
+
+PY_FORMATS = [
+	("csv", "csv_queries", '"csv"'),
+	("json", "json_queries", '"json"'),
+	("rdf", "rdf_queries", '"rdf" / "oxigraph_store"'),
+]
+
+
+def write_query_files(bench_dir: Path) -> int:
+	queries_dir = bench_dir / "queries"
+	written = 0
+	for fmt, module, subpath in PY_FORMATS:
+		fmt_dir = queries_dir / fmt
+		fmt_dir.mkdir(parents=True, exist_ok=True)
+		for n in range(1, 8):
+			text = WRAPPER.format(n=n, fmt=fmt, module=module, subpath=subpath)
+			(fmt_dir / f"q{n}.py").write_text(text, encoding="ascii", newline="\n")
+			written += 1
+	for fmt, sql in (("sqlite", SQLITE_SQL), ("pg", PG_SQL)):
+		fmt_dir = queries_dir / fmt
+		fmt_dir.mkdir(parents=True, exist_ok=True)
+		for n, text in sql.items():
+			(fmt_dir / f"q{n}.sql").write_text(text, encoding="ascii", newline="\n")
+			written += 1
+	rdf_dir = queries_dir / "rdf"
+	for n, keys in RDF_SPARQL_FILES.items():
+		text = "\n\n".join(f"# retrieval query: {key}\n{rdf_queries.SPARQL[key]}" for key in keys)
+		(rdf_dir / f"q{n}.sparql").write_text(text + "\n", encoding="ascii", newline="\n")
+		written += 1
+	logger.info("wrote %d query files under %s", written, queries_dir)
+	return written
+
+
+def run_sqlite(db_path: Path) -> dict[int, list[tuple]]:
+	conn = sqlite3.connect(db_path)
+	try:
+		return {n: [tuple(r) for r in conn.execute(sql).fetchall()] for n, sql in SQLITE_SQL.items()}
+	finally:
+		conn.close()
+
+
+def run_pg(dsn: str) -> dict[int, list[tuple]]:
+	results: dict[int, list[tuple]] = {}
+	with psycopg.connect(dsn) as conn:
+		for n, sql in PG_SQL.items():
+			results[n] = [tuple(r) for r in conn.execute(sql).fetchall()]
+			logger.info("pg q%d: %d rows", n, len(results[n]))
+	return results
+
+
+def main() -> int:
+	args = parse_task_args("Task 08: define benchmark queries and canonical expected results")
+	out_root: Path = args.out_root
+	bench_dir = out_root / "bench"
+	try:
+		check_dependencies(out_root, DEPENDS_ON)
+		remove_stale_marker(out_root, TASK_ID)
+		files_written = write_query_files(bench_dir)
+
+		logger.info("computing canonical results on PostgreSQL")
+		pg_rows = run_pg(get_dsn())
+		logger.info("cross-validating against SQLite")
+		sqlite_rows = run_sqlite(out_root / "sqlite" / "tribo.db")
+		for n in range(1, 8):
+			diff = compare(pg_rows[n], sqlite_rows[n])
+			if diff:
+				raise ValueError(f"q{n}: SQLite disagrees with PostgreSQL: {diff}")
+			exact = EXACT_ROW_COUNTS.get(n)
+			if exact is not None and len(pg_rows[n]) != exact:
+				raise ValueError(f"q{n}: {len(pg_rows[n])} rows, expected {exact}")
+			if not pg_rows[n]:
+				raise ValueError(f"q{n}: empty result set")
+
+		logger.info("spot-checking csv/json/rdf on Q1")
+		for label, rows in (
+			("csv", csv_queries.q1(out_root / "csv")),
+			("json", json_queries.q1(out_root / "json")),
+			("rdf", rdf_queries.q1(out_root / "rdf" / "oxigraph_store")),
+		):
+			diff = compare(pg_rows[1], rows)
+			if diff:
+				raise ValueError(f"q1 via {label} disagrees with canonical: {diff}")
+
+		expected_dir = bench_dir / "expected"
+		expected_dir.mkdir(parents=True, exist_ok=True)
+		checksums = {}
+		for n in range(1, 8):
+			rows = sort_rows(pg_rows[n])
+			(expected_dir / f"q{n}.rows.csv").write_text(serialize(rows), encoding="ascii", newline="\n")
+			checksums[n] = sha256_of(rows)
+			(expected_dir / f"q{n}.sha256").write_text(checksums[n] + "\n", encoding="ascii", newline="\n")
+
+		entries = {"query_files": files_written}
+		for n in range(1, 8):
+			entries[f"q{n}_rows"] = len(pg_rows[n])
+			entries[f"q{n}_sha256"] = checksums[n][:12]
+		entries.update(process_metrics())
+		marker_path = write_marker(out_root, TASK_ID, entries)
+	except Exception:
+		logger.critical("task %s failed", TASK_ID, exc_info=True)
+		return 1
+
+	counts = ", ".join(f"q{n}={len(pg_rows[n])}" for n in range(1, 8))
+	print(f"task 08 ok: {files_written} query files, canonical results validated (pg == sqlite, 1e-9)")
+	print(f"rows: {counts}")
+	print(f"expected: {bench_dir / 'expected'} (q<N>.rows.csv + q<N>.sha256)")
+	print(f"marker: {marker_path}")
+	return 0
+
+
+if __name__ == "__main__":
+	sys.exit(main())