diff --git a/README.md b/README.md index 84ab655..48689b4 100644 --- a/README.md +++ b/README.md @@ -61,7 +61,7 @@ machine (no parallel work). Every task writes a completion marker | 02 Process flow diagrams | `make_process_flow.py` | `out/report/process_flow.md`, `out/report/diagrams/` | implemented | | 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` | planned | +| 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 | @@ -142,6 +142,7 @@ measurement counts. See docs/ specs/ task specifications 00-11 (WHAT to build) rules/ binding conventions (HOW work is done) + research/ measured design studies (e.g. FK key-schema study) examples/ imported reference materials (not binding) out/ ALL generated artifacts (git-ignored, reproducible): config/ csv/ json/ sqlite/ pg/ rdf/ bench/ report/ .done/ @@ -150,6 +151,7 @@ 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 requirements.txt closed dependency list (docs/rules/code-python-style.md) ``` diff --git a/common/corpus.py b/common/corpus.py new file mode 100644 index 0000000..e1e5e62 --- /dev/null +++ b/common/corpus.py @@ -0,0 +1,73 @@ +"""Reading the CSV corpus with MANIFEST validation (converters, tasks 04-07). + +Every file consumed from ./out/csv/ is verified against MANIFEST.csv (byte +size, sha256, row count) before its content is trusted - a mismatch aborts +the converter (docs/rules/test-pipeline-validation.md section 2). +""" + +from __future__ import annotations + +import hashlib +import logging +import re +from pathlib import Path + +logger = logging.getLogger(__name__) + +MANIFEST_HEADER = "path,rows,bytes,sha256" + +INT_RE = re.compile(r"^-?[0-9]+$") +FLOAT_RE = re.compile(r"^-?([0-9]+\.[0-9]*|\.[0-9]+|[0-9]+)([eE][+-]?[0-9]+)?$") + + +class ValidationError(RuntimeError): + pass + + +def coerce(s: str) -> int | float | str: + if INT_RE.match(s): + return int(s) + if FLOAT_RE.match(s): + return float(s) + return s + + +class CorpusReader: + def __init__(self, csv_root: Path) -> None: + self.csv_root = csv_root + self.manifest = self._load_manifest() + + def _load_manifest(self) -> dict[str, tuple[int, int, str]]: + manifest: dict[str, tuple[int, int, str]] = {} + with open(self.csv_root / "MANIFEST.csv", encoding="ascii") as fh: + header = fh.readline().strip() + if header != MANIFEST_HEADER: + raise ValidationError(f"MANIFEST.csv: unexpected header {header!r}") + for line in fh: + path, rows, nbytes, sha = line.strip().split(",") + manifest[path] = (int(rows), int(nbytes), sha) + logger.info("manifest loaded: %d entries", len(manifest)) + return manifest + + def read(self, relpath: str) -> tuple[list[str], list[list[str]]]: + """Return (header, raw string rows), verified against MANIFEST.""" + data = (self.csv_root / relpath).read_bytes() + exp_rows, exp_bytes, exp_sha = self.manifest[relpath] + if len(data) != exp_bytes or hashlib.sha256(data).hexdigest() != exp_sha: + raise ValidationError(f"{relpath}: bytes/sha256 mismatch vs MANIFEST") + lines = data.decode("ascii").split("\n") + header = lines[0].split(",") + rows = [ln.split(",") for ln in lines[1:] if ln] + if len(rows) != exp_rows: + raise ValidationError(f"{relpath}: {len(rows)} rows != manifest {exp_rows}") + return header, rows + + def dicts(self, relpath: str) -> list[dict]: + """Rows as dicts: lowercased CSV headers as keys, numerics coerced, + empty cells omitted (docs/rules/data-naming-units.md section 4).""" + header, rows = self.read(relpath) + keys = [h.lower() for h in header] + return [ + {k: coerce(v) for k, v in zip(keys, row) if v != ""} + for row in rows + ] diff --git a/common/pipeline.py b/common/pipeline.py new file mode 100644 index 0000000..89b83a1 --- /dev/null +++ b/common/pipeline.py @@ -0,0 +1,59 @@ +"""Task-orchestration helpers: CLI shape, dependency markers, completion +markers, lab config loading (docs/rules/build-pipeline-tasks.md, +docs/rules/code-python-style.md section 3). +""" + +from __future__ import annotations + +import argparse +import logging +import sys +from pathlib import Path + +import yaml + +logger = logging.getLogger(__name__) + +REPO_ROOT = Path(__file__).resolve().parent.parent + + +def parse_task_args(description: str) -> argparse.Namespace: + parser = argparse.ArgumentParser(description=description) + parser.add_argument("--log-level", default="INFO", choices=["DEBUG", "INFO", "WARNING", "ERROR"]) + parser.add_argument("--out-root", type=Path, default=REPO_ROOT / "out", help="artifact tree root (default: ./out)") + args = parser.parse_args() + logging.basicConfig( + level=args.log_level, + stream=sys.stderr, + format="%(asctime)s %(levelname)-7s %(name)s: %(message)s", + ) + return args + + +def check_dependencies(out_root: Path, deps: list[str]) -> None: + for dep in deps: + marker = out_root / ".done" / f"{dep}.ok" + if not marker.exists(): + raise FileNotFoundError(f"dependency marker missing: {marker} - run task {dep} first") + + +def remove_stale_marker(out_root: Path, task_id: str) -> Path: + marker = out_root / ".done" / f"{task_id}.ok" + if marker.exists(): + logger.info("re-run: removing stale marker %s", marker) + marker.unlink() + return marker + + +def write_marker(out_root: Path, task_id: str, entries: dict) -> Path: + marker = out_root / ".done" / f"{task_id}.ok" + lines = [f"task: '{task_id}'", "status: ok"] + [f"{k}: {v}" for k, v in entries.items()] + marker.parent.mkdir(parents=True, exist_ok=True) + with open(marker, "w", encoding="utf-8", newline="\n") as fh: + fh.write("\n".join(lines) + "\n") + return marker + + +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) diff --git a/common/track_summary.py b/common/track_summary.py new file mode 100644 index 0000000..0c7eb43 --- /dev/null +++ b/common/track_summary.py @@ -0,0 +1,48 @@ +"""Per-track steady-state summary (docs/rules/db-sql-schema.md section 5). + +Both engines must produce identical values: SQLite calls this function +during load (task 05); the PostgreSQL materialized view (task 06) +implements the same three steps in SQL. Plain sequential float arithmetic +keeps results within the 1e-9 checksum tolerance across implementations. +""" + +from __future__ import annotations + +import math + +TAIL_START_CYCLE = 501 # steady-state proxy window: cycles 501..N + + +def summarize_track(cofs: list[float]) -> tuple[float, float, int]: + """Return (cof_ss_mean, cof_ss_std, run_in_cycles) for cofs[i] = cof at cycle i+1.""" + n = len(cofs) + tail = cofs[TAIL_START_CYCLE - 1:] + m = _mean(tail) + s = _stddev_pop(tail, m) + + run_in = TAIL_START_CYCLE - 1 # fallback when no cycle enters the 2-sigma band + band = 2.0 * s + for c in range(1, n + 1): + if abs(cofs[c - 1] - m) < band: + run_in = c + break + + ss = cofs[run_in:] # cycles strictly greater than run_in + ss_mean = _mean(ss) + ss_std = _stddev_pop(ss, ss_mean) + return ss_mean, ss_std, run_in + + +def _mean(values: list[float]) -> float: + total = 0.0 + for v in values: + total += v + return total / len(values) + + +def _stddev_pop(values: list[float], mean: float) -> float: + acc = 0.0 + for v in values: + d = v - mean + acc += d * d + return math.sqrt(acc / len(values)) diff --git a/convert_json.py b/convert_json.py index 87ab865..22ad48b 100644 --- a/convert_json.py +++ b/convert_json.py @@ -17,23 +17,26 @@ Appends both sizes to ./out/bench/storage_sizes.csv and writes from __future__ import annotations -import argparse -import hashlib import json import logging -import re import shutil import sys from pathlib import Path -import yaml from rdflib import Graph +from common.corpus import CorpusReader, ValidationError +from common.pipeline import ( + check_dependencies, + load_lab_config, + parse_task_args, + remove_stale_marker, + write_marker, +) from common.storage_sizes import update_storage_sizes logger = logging.getLogger(__name__) -REPO_ROOT = Path(__file__).resolve().parent TASK_ID = "04" DEPENDS_ON = ["01", "03"] @@ -60,9 +63,6 @@ CONTEXT = { "started_at": {"@id": "prov:startedAtTime"}, } -INT_RE = re.compile(r"^-?[0-9]+$") -FLOAT_RE = re.compile(r"^-?([0-9]+\.[0-9]*|\.[0-9]+|[0-9]+)([eE][+-]?[0-9]+)?$") - MEASUREMENTS = [ # (csv file, json key, @type, instrument id, points key) ("xrf_map.csv", "xrf", "CompositionMeasurement", "m4_tornado", "points"), @@ -71,61 +71,19 @@ MEASUREMENTS = [ ] -class ValidationError(RuntimeError): - pass - - -def coerce(s: str) -> int | float | str: - if INT_RE.match(s): - return int(s) - if FLOAT_RE.match(s): - return float(s) - return s - - class JsonConverter: def __init__(self, cfg: dict, csv_root: Path, json_root: Path) -> None: self.cfg = cfg - self.csv_root = csv_root self.json_root = json_root - self.manifest = self._load_manifest() + self.reader = CorpusReader(csv_root) + self.manifest = self.reader.manifest self.graph: list[dict] = [] # hybrid metadata entities self.full_bytes = 0 self.full_files = 0 self.counts = {"coupons": 0, "tracks": 0, "cycle_rows": 0, "loop_rows": 0, "xrf_rows": 0} - def _load_manifest(self) -> dict[str, tuple[int, int, str]]: - manifest: dict[str, tuple[int, int, str]] = {} - with open(self.csv_root / "MANIFEST.csv", encoding="ascii") as fh: - header = fh.readline().strip() - if header != "path,rows,bytes,sha256": - raise ValidationError(f"MANIFEST.csv: unexpected header {header!r}") - for line in fh: - path, rows, nbytes, sha = line.strip().split(",") - manifest[path] = (int(rows), int(nbytes), sha) - logger.info("manifest loaded: %d entries", len(manifest)) - return manifest - - def read_csv(self, relpath: str) -> tuple[list[str], list[list[str]]]: - """Read one corpus file, verifying bytes and row count against MANIFEST.""" - data = (self.csv_root / relpath).read_bytes() - exp_rows, exp_bytes, exp_sha = self.manifest[relpath] - if len(data) != exp_bytes or hashlib.sha256(data).hexdigest() != exp_sha: - raise ValidationError(f"{relpath}: bytes/sha256 mismatch vs MANIFEST") - lines = data.decode("ascii").split("\n") - header = lines[0].split(",") - rows = [ln.split(",") for ln in lines[1:] if ln] - if len(rows) != exp_rows: - raise ValidationError(f"{relpath}: {len(rows)} rows != manifest {exp_rows}") - return header, rows - def rows_as_dicts(self, relpath: str) -> list[dict]: - header, rows = self.read_csv(relpath) - keys = [h.lower() for h in header] - return [ - {k: coerce(v) for k, v in zip(keys, row) if v != ""} - for row in rows - ] + return self.reader.dicts(relpath) def source_ref(self, relpath: str) -> dict: rows, _, sha = self.manifest[relpath] @@ -285,54 +243,17 @@ def validate_sample(paths: list[Path]) -> None: logger.info("sample %s: valid JSON, %d triples via rdflib", path.name, len(g)) -def check_dependencies(out_root: Path) -> None: - for dep in DEPENDS_ON: - marker = out_root / ".done" / f"{dep}.ok" - if not marker.exists(): - raise FileNotFoundError(f"dependency marker missing: {marker} - run task {dep} first") - - -def write_marker(marker_path: Path, conv: JsonConverter, hybrid_bytes: int) -> None: - lines = [ - f"task: '{TASK_ID}'", - "status: ok", - f"full_files: {conv.full_files}", - f"full_bytes: {conv.full_bytes}", - f"hybrid_bytes: {hybrid_bytes}", - f"coupons: {conv.counts['coupons']}", - f"tracks: {conv.counts['tracks']}", - f"cycle_rows: {conv.counts['cycle_rows']}", - f"loop_points: {conv.counts['loop_rows']}", - ] - marker_path.parent.mkdir(parents=True, exist_ok=True) - with open(marker_path, "w", encoding="utf-8", newline="\n") as fh: - fh.write("\n".join(lines) + "\n") - - def main() -> int: - parser = argparse.ArgumentParser(description="Task 04: convert the CSV corpus to JSON-LD") - parser.add_argument("--log-level", default="INFO", choices=["DEBUG", "INFO", "WARNING", "ERROR"]) - parser.add_argument("--out-root", type=Path, default=REPO_ROOT / "out", help="artifact tree root (default: ./out)") - args = parser.parse_args() - logging.basicConfig( - level=args.log_level, - stream=sys.stderr, - format="%(asctime)s %(levelname)-7s %(name)s: %(message)s", - ) - + args = parse_task_args("Task 04: convert the CSV corpus to JSON-LD") out_root: Path = args.out_root json_root = out_root / "json" - marker_path = out_root / ".done" / f"{TASK_ID}.ok" try: - check_dependencies(out_root) - if marker_path.exists(): - logger.info("re-run: removing stale marker %s", marker_path) - marker_path.unlink() + check_dependencies(out_root, DEPENDS_ON) + remove_stale_marker(out_root, TASK_ID) if json_root.exists(): logger.info("re-run: removing previous output %s", json_root) shutil.rmtree(json_root) - with open(out_root / "config" / "lab_config.yaml", encoding="utf-8") as fh: - cfg = yaml.safe_load(fh) + cfg = load_lab_config(out_root) conv = JsonConverter(cfg, out_root / "csv", json_root) conv.convert_flat() @@ -344,7 +265,15 @@ def main() -> int: json_root / "hybrid" / "dataset.jsonld", ]) update_storage_sizes(out_root / "bench", "json", [("full", conv.full_bytes), ("hybrid", hybrid_bytes)]) - write_marker(marker_path, conv, hybrid_bytes) + marker_path = write_marker(out_root, TASK_ID, { + "full_files": conv.full_files, + "full_bytes": conv.full_bytes, + "hybrid_bytes": hybrid_bytes, + "coupons": conv.counts["coupons"], + "tracks": conv.counts["tracks"], + "cycle_rows": conv.counts["cycle_rows"], + "loop_points": conv.counts["loop_rows"], + }) except Exception: logger.critical("task %s failed", TASK_ID, exc_info=True) return 1 diff --git a/convert_sqlite.py b/convert_sqlite.py new file mode 100644 index 0000000..09e63b8 --- /dev/null +++ b/convert_sqlite.py @@ -0,0 +1,423 @@ +"""Task 05 - Convert the CSV corpus to SQLite. + +Builds ./out/sqlite/tribo.db: variant B key schema (composite natural PKs +on bulk tables, WITHOUT ROWID, enforced foreign keys - see +docs/research/Tribology_FK_Architecture.html), Q1-Q7 indexes created after +the load, track_summary precomputed. Every consumed CSV is verified +against MANIFEST.csv. Appends the db size to storage_sizes.csv and writes +./out/.done/05.ok. Spec: docs/specs/05_convert_sqlite.md. +""" + +from __future__ import annotations + +import logging +import sqlite3 +import sys +from pathlib import Path + +from common.corpus import CorpusReader, ValidationError +from common.pipeline import ( + check_dependencies, + load_lab_config, + parse_task_args, + remove_stale_marker, + write_marker, +) +from common.storage_sizes import update_storage_sizes +from common.track_summary import summarize_track + +logger = logging.getLogger(__name__) + +TASK_ID = "05" +DEPENDS_ON = ["01", "03"] + +BATCH_ROWS = 50000 # rows per executemany transaction, spec 05 +LOAD_PRAGMAS = [ + "PRAGMA journal_mode=OFF", + "PRAGMA synchronous=OFF", + "PRAGMA cache_size=-262144", + "PRAGMA foreign_keys=ON", +] + +DDL = [ + """CREATE TABLE instruments ( + instrument_id INTEGER PRIMARY KEY, + instrument_code TEXT NOT NULL UNIQUE, + name TEXT NOT NULL, + role TEXT NOT NULL + )""", + """CREATE TABLE batches ( + batch_id INTEGER PRIMARY KEY, + batch_code TEXT NOT NULL UNIQUE, + 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 TEXT NOT NULL + )""", + """CREATE TABLE runs ( + run_id INTEGER PRIMARY KEY, + run_code TEXT NOT NULL UNIQUE, + environment TEXT NOT NULL, + date TEXT NOT NULL, + plates INTEGER NOT NULL, + operator TEXT NOT NULL + )""", + """CREATE TABLE wafers ( + wafer_id INTEGER PRIMARY KEY, + wafer_code TEXT NOT NULL UNIQUE, + batch_id INTEGER NOT NULL REFERENCES batches(batch_id), + wafer_index INTEGER NOT NULL, + deposition_date TEXT NOT NULL, + coupons INTEGER NOT NULL, + friction_coupons INTEGER NOT NULL, + reserve_coupons INTEGER NOT NULL + )""", + """CREATE TABLE coupons ( + coupon_id INTEGER PRIMARY KEY, + coupon_code TEXT NOT NULL UNIQUE, + wafer_id INTEGER NOT NULL REFERENCES wafers(wafer_id), + batch_id INTEGER NOT NULL REFERENCES batches(batch_id), + grid_row INTEGER NOT NULL, + grid_col INTEGER NOT NULL, + thickness_um REAL NOT NULL, + ra_nm REAL NOT NULL, + au_wtpct_mean REAL NOT NULL, + run_id INTEGER REFERENCES runs(run_id), + plate INTEGER, + probe INTEGER, + square INTEGER + )""", + """CREATE TABLE tracks ( + track_id INTEGER PRIMARY KEY, + track_code TEXT NOT NULL UNIQUE, + coupon_id INTEGER NOT NULL REFERENCES coupons(coupon_id), + run_id INTEGER NOT NULL REFERENCES runs(run_id), + environment TEXT NOT NULL, + load_mn REAL NOT NULL, + stroke_mm REAL NOT NULL, + speed_mm_s REAL NOT NULL, + counterface_id TEXT NOT NULL, + started_at TEXT NOT NULL + )""", + """CREATE TABLE simtra_profiles ( + batch_id INTEGER NOT NULL REFERENCES batches(batch_id), + row_no INTEGER NOT NULL, + angle_deg REAL NOT NULL, + energy_ev REAL NOT NULL, + pt_flux REAL NOT NULL, + au_flux REAL NOT NULL, + PRIMARY KEY (batch_id, row_no) + ) WITHOUT ROWID""", + """CREATE TABLE xrf_points ( + coupon_id INTEGER NOT NULL REFERENCES coupons(coupon_id), + grid_x INTEGER NOT NULL, + grid_y INTEGER NOT NULL, + pt_wtpct REAL NOT NULL, + au_wtpct REAL NOT NULL, + PRIMARY KEY (coupon_id, grid_x, grid_y) + ) WITHOUT ROWID""", + """CREATE TABLE profilometry_points ( + coupon_id INTEGER NOT NULL REFERENCES coupons(coupon_id), + grid_x INTEGER NOT NULL, + grid_y INTEGER NOT NULL, + thickness_um REAL NOT NULL, + PRIMARY KEY (coupon_id, grid_x, grid_y) + ) WITHOUT ROWID""", + """CREATE TABLE nanoindentation ( + coupon_id INTEGER NOT NULL REFERENCES coupons(coupon_id), + indent_id INTEGER NOT NULL, + x_um REAL NOT NULL, + y_um REAL NOT NULL, + hardness_gpa REAL NOT NULL, + reduced_modulus_gpa REAL NOT NULL, + max_load_mn REAL NOT NULL, + PRIMARY KEY (coupon_id, indent_id) + ) WITHOUT ROWID""", + """CREATE TABLE afm ( + coupon_id INTEGER PRIMARY KEY REFERENCES coupons(coupon_id), + ra_nm REAL NOT NULL, + rq_nm REAL NOT NULL, + image_file TEXT NOT NULL + )""", + """CREATE TABLE friction_cycles ( + track_id INTEGER NOT NULL REFERENCES tracks(track_id), + cycle INTEGER NOT NULL, + cof REAL NOT NULL, + PRIMARY KEY (track_id, cycle) + ) WITHOUT ROWID""", + """CREATE TABLE friction_loop_points ( + track_id INTEGER NOT NULL REFERENCES tracks(track_id), + cycle INTEGER NOT NULL, + pt INTEGER NOT NULL, + position_um REAL NOT NULL, + friction_force_mn REAL NOT NULL, + PRIMARY KEY (track_id, cycle, pt) + ) WITHOUT ROWID""", + """CREATE TABLE wear ( + track_id INTEGER PRIMARY KEY REFERENCES tracks(track_id), + wear_volume_um3 REAL NOT NULL, + k_archard REAL NOT NULL, + sliding_distance_m REAL NOT NULL + )""", + """CREATE TABLE track_summary ( + track_id INTEGER PRIMARY KEY REFERENCES tracks(track_id), + cof_ss_mean REAL NOT NULL, + cof_ss_std REAL NOT NULL, + run_in_cycles INTEGER NOT NULL + )""", +] + +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 +] + +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) + 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] = {} + + 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 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 --- + + def finalize(self) -> None: + for ddl in INDEX_DDL: + self.conn.execute(ddl) + self.conn.commit() + self.conn.execute("PRAGMA journal_mode=WAL") + self.conn.execute("ANALYZE") + 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"], + } + counts: dict[str, int] = {} + for table, exp in expected.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"]: + 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) + return counts + + +def main() -> int: + args = parse_task_args("Task 05: convert the CSV corpus to SQLite") + out_root: Path = args.out_root + db_path = out_root / "sqlite" / "tribo.db" + try: + check_dependencies(out_root, DEPENDS_ON) + remove_stale_marker(out_root, TASK_ID) + for stale in (db_path, db_path.with_suffix(".db-wal"), db_path.with_suffix(".db-shm")): + if stale.exists(): + logger.info("re-run: removing %s", stale) + stale.unlink() + cfg = load_lab_config(out_root) + + loader = SqliteLoader(cfg, out_root / "csv", db_path) + loader.load_flat() + loader.load_hierarchy() + loader.finalize() + counts = loader.validate() + 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, { + "db_file": db_path.as_posix(), + "db_bytes": db_bytes, + "tables": len(counts), + "total_rows": sum(counts.values()), + "friction_cycles": counts["friction_cycles"], + "friction_loop_points": counts["friction_loop_points"], + "tracks": counts["tracks"], + "track_summary": counts["track_summary"], + }) + except Exception: + logger.critical("task %s failed", TASK_ID, exc_info=True) + return 1 + + print(f"task 05 ok: {db_path} ({db_bytes / (1024 * 1024):.1f} MiB, {len(counts)} tables, {sum(counts.values())} rows)") + print(f"bulk rows: cycles={counts['friction_cycles']} loop_points={counts['friction_loop_points']} " + f"xrf={counts['xrf_points']} track_summary={counts['track_summary']}") + print(f"marker: {marker_path}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/docs/research/Tribology_FK_Architecture.html b/docs/research/Tribology_FK_Architecture.html new file mode 100644 index 0000000..1646f48 --- /dev/null +++ b/docs/research/Tribology_FK_Architecture.html @@ -0,0 +1,571 @@ + + + + + +Foreign-Key Architecture Study - Tribology Lab Data Storage + + + + + + +
+ +
+ +
+ Foreign-Key Architecture Study + Tribology lab data storage · bulk-table key design +
+
+ SQLite · 4.32M rows measured + Measured 2026-07-11 +
+
+ +
+ + + +
+ + +
+

Overview

+

Three candidate key schemas for the two largest measurement tables (friction_cycles, 1.44M rows and friction_loop_points, 2.88M rows) were built from the same simulated corpus and measured for disk footprint, load time and six read patterns. In every variant the relations between tables are single-column integer foreign keys (track_id, coupon_id); the variants differ only in how a bulk row is identified inside its own table.

+ +
+
Disk, variant C vs A
+74%
190.2 vs 109.1 MiB
+
GROUP BY scan, C vs A
+59%
135.9 vs 85.7 ms (Q5 pattern)
+
Enforced FK cost (B)
+0.5 s
load only; disk and reads = A
+
Point lookups
±2%
identical across A / B / C
+
+ +
+

The three variants

+
    +
  • Variant A - composite natural PK. Row identity of a bulk row is its natural composite key (track_id, cycle), stored as a clustered WITHOUT ROWID table. FK columns are plain integers, validated by the loader against the corpus manifest (no engine constraint).
    • +
  • Variant B - composite natural PK + enforced FK. Same physical layout as A, plus REFERENCES constraints with PRAGMA foreign_keys=ON, so the engine itself guarantees referential integrity.
    • +
  • Variant C - surrogate key on every table. Classic style: each table gets a single-column auto-increment key named <singular>_id (friction_cycle_id, never a bare id). Correctness then requires an additional UNIQUE(track_id, cycle) index - a second B-tree.
    • +
+
Referential integrity is not what differs between the variants: every relation is a single integer FK in all three. What differs is the number of B-trees the engine must store and traverse per bulk table: one (A, B) or two (C).
+
+
+ + +
+

Sample Dataset

+

All measurements ran against the real generated corpus of the evaluation pipeline - deterministic, physically plausible tribology data (seed 20260711). The two bulk tables and their parent dimension were loaded in full.

+
+ + + + + + + +
TableGrainRows loadedColumns (beyond keys)Source files
tracksone friction track (3 per tested coupon)1,440track_code, coupon_id, run_id, environment, load_mN, ...track_info.csv × 1,440
friction_cyclesone reciprocating cycle of one track1,440,000cofcof_vs_cycle.csv × 1,440
friction_loop_pointsone position-resolved loop point (10 loops × 200 pts per track)2,880,000position_um, friction_force_mNfriction_loops.csv × 1,440
+
+
+

Corpus context

+

The corpus models a Pt-Au coating study: 4 deposition batches → 12 wafers → 588 coupons (480 friction-tested) → 1,440 tracks → 1.44M cycle records. Generated size: 141.8 MiB of CSV; the production target for extrapolation is 600 GB - 6 TB, which is why per-row bytes on the bulk tables matter.

+
+
+ + +
+

Variant A · Composite Natural Primary Key baseline

+

The bulk row is identified by what it physically is: cycle n of track t. The composite PK doubles as the clustering order, so the table itself is the only B-tree. FK columns are unenforced integers; integrity is guaranteed by row-count and checksum validation against the corpus manifest.

+ +
+
+ + + + + 1 + 1 + ∞ + ∞ + +
+
tracksdimension
+
track_idPK
+
track_codeUK
+
coupon_idFK
+
run_idFK
+
environmenttext
+
load_mnreal
+
+
+
friction_cycles1.44M rows
+
track_idPK · ref
+
cyclePK
+
cofreal
+
+
+
friction_loop_points2.88M rows
+
track_idPK · ref
+
cyclePK
+
ptPK
+
position_umreal
+
friction_force_mnreal
+
+
+ wear and track_summary follow the same pattern: single-column PK = FK track_id (1:1 with tracks).

+ ref = integer reference to tracks(track_id), validated by the loader, no engine constraint. +
+
+ 
CREATE TABLE friction_cycles (
+  track_id INTEGER NOT NULL,   -- reference to tracks(track_id), loader-validated
+  cycle    INTEGER NOT NULL,
+  cof      REAL    NOT NULL,
+  PRIMARY KEY (track_id, cycle)
+) WITHOUT ROWID;              -- clustered: the PK is the table's only B-tree
+
+
+ + +
+

Variant B · Composite PK + Enforced Foreign Keys recommended

+

Physically identical to Variant A - same clustered layout, same single B-tree per bulk table. The only change: FK columns carry real REFERENCES constraints and every connection runs PRAGMA foreign_keys=ON, so the DBMS itself rejects an orphan row. Measured cost: +0.5 s of load time on 4.32M rows; zero cost on disk and reads.

+ +
+
+ + + + + 1 + 1 + ∞ + ∞ + +
+
tracksdimension
+
track_idPK
+
track_codeUK
+
coupon_idFK
+
run_idFK
+
environmenttext
+
load_mnreal
+
+
+
friction_cycles1.44M rows
+
track_idPK
+
track_id → tracksFK ✓
+
cyclePK
+
cofreal
+
+
+
friction_loop_points2.88M rows
+
track_idPK
+
track_id → tracksFK ✓
+
cyclePK
+
ptPK
+
position_um, friction_force_mnreal
+
+
+ PRAGMA foreign_keys = ON on every connection - SQLite silently ignores FK clauses without it.

+ FK checks are lookups into the already-indexed parent PK, which is why the measured overhead is only ~6% of load time. +
+
+ 
PRAGMA foreign_keys = ON;
+
+CREATE TABLE friction_cycles (
+  track_id INTEGER NOT NULL REFERENCES tracks(track_id),
+  cycle    INTEGER NOT NULL,
+  cof      REAL    NOT NULL,
+  PRIMARY KEY (track_id, cycle)
+) WITHOUT ROWID;
+
+
+ + +
+

Variant C · Surrogate <singular>_id on Every Table +74% disk

+

The uniform classic design: every table, including the bulk fact tables, gets a single-column auto-increment key (friction_cycle_id, friction_loop_point_id - by convention always <table name in singular>_id, never a bare id). In SQLite the id column itself is nearly free (it aliases the rowid), but correctness still demands UNIQUE(track_id, cycle) - and that unique index is a second B-tree holding a copy of almost every column.

+ +
+
+ + + + + 1 + 1 + ∞ + ∞ + +
+
tracksdimension
+
track_idPK
+
track_codeUK
+
coupon_idFK
+
run_idFK
+
environmenttext
+
load_mnreal
+
+
+
friction_cycles1.44M rows
+
friction_cycle_idPK
+
track_id → tracksFK ✓
+
cycleint
+
cofreal
+
UNIQUE (track_id, cycle)2nd B-tree
+
+
+
friction_loop_points2.88M rows
+
friction_loop_point_idPK
+
track_id → tracksFK ✓
+
cycleint
+
ptint
+
position_umreal
+
friction_force_mnreal
+
UNIQUE (track_id, cycle, pt)2nd B-tree
+
+
+ The surrogate key is never referenced by any other table - no row in the schema points at an individual cycle.

+ The UNIQUE index is mandatory for correctness (it prevents duplicate cycles), so its cost cannot be avoided in this variant. +
+
+ 
CREATE TABLE friction_cycles (
+  friction_cycle_id INTEGER PRIMARY KEY,        -- rowid alias; unused by any FK
+  track_id INTEGER NOT NULL REFERENCES tracks(track_id),
+  cycle    INTEGER NOT NULL,
+  cof      REAL    NOT NULL,
+  UNIQUE (track_id, cycle)                      -- required; duplicates the table
+);
+
+
+ + +
+

Benchmark Method

+

Identical protocol for all three variants; only the DDL differs.

+
+
    +
  • Data: the full generated corpus subset - tracks (1,440), friction_cycles (1,440,000), friction_loop_points (2,880,000) - read from the same CSV files in the same order for every variant.
    • +
  • Load: SQLite, PRAGMA journal_mode=OFF, synchronous=OFF, cache_size=-262144; inserts via executemany in 50k-row batches; VACUUM before measuring the file size.
    • +
  • Reads: warm OS page cache, best of 3 runs per pattern; fixed deterministic probe sets (39 tracks for range reads, 150 (track_id, cycle) pairs for point lookups); result-set sizes asserted, so every variant answers the identical question.
    • +
  • Caveat: at 150 MB everything fits in RAM, which flatters variant C - at the 600 GB production target the working set exceeds RAM, and twice the pages means twice the cache pressure and disk reads. Warm-cache deltas are therefore a lower bound.
    • +
+
+
+ + +
+

Results · Storage & Load

+

File size after VACUUM; load time includes CSV parsing (identical work in all variants).

+ +
+ + + + + + + +
VariantBulk-table keysLoad time (s)DB size (MiB)Size vs A
A · composite PKPK (track_id, cycle), unenforced refs7.5109.1baseline
B · composite PK + FKPK (track_id, cycle), REFERENCES enforced8.0109.1±0%
C · surrogate idfriction_cycle_id PK + UNIQUE(track_id, cycle)8.8190.2+74%
+
+ +
+

Database size after VACUUM (MiB, scale 0-200)

+
+
Variant Acomposite PK, unenforced refs
109.1 MiB
+
Variant Bcomposite PK + enforced FK
109.1 MiB
+
Variant Csurrogate id + UNIQUE index
190.2 MiB
+
+

Load time, 4.32M rows (s, scale 0-10)

+
+
Variant A
7.5 s
+
Variant B
8.0 s
+
Variant C
8.8 s
+
+
Extrapolated to the 600 GB production target, the +74% of variant C on the two hottest tables translates into hundreds of gigabytes of extra disk and a doubled page count competing for the same RAM cache.
+
+
+ + +
+

Results · Read Performance

+

Warm OS cache, best of 3 runs. Q1 / Q5 refer to the benchmark scenarios of the evaluation pipeline.

+ +
+ + + + + + + + + + +
Read patternUnitABCC vs A
Point lookup of one row by (track_id, cycle)µs/op24.823.624.3~0%
Q1: all 1,000 cycles of one trackms/track0.4730.4850.533+13%
All 2,000 loop points of one trackms/track1.2671.2991.336+5%
Full scan + AVG over 1.44M cyclesms64.964.970.7+9%
AVG with GROUP BY track_id over 1.44M cycles (Q5 pattern)ms85.789.1135.9+59%
Full scan over 2.88M loop pointsms152.5150.8165.4+8%
+
+ +
+

Relative read time, variant A = 100% (lower is better)

+
+ A · composite PK + B · + enforced FK + C · surrogate id + UNIQUE +
+ + + + + + 0% + 50% + 100% + 150% + + + + + + Point lookup + + + + + Q1: track cycles + + + + + Track loop points + + + + + Full scan + AVG + + + + + +59% + GROUP BY track (Q5) + + + + + Loop points scan + + dashed line = variant A baseline (100%) + +
+
+ + +
+

Why It Differs

+

The whole difference reduces to one thing: how many B-trees per bulk table, and in what order the rows physically live.

+ +
+
+ + + extra hop + +
+
Variants A / B · one B-tree
+
clustered table keyed by (track_id, cycle)
+
descent → leaf → sequential range scan
+
rows of one track are physically adjacent
+
+
+
Variant C · B-tree 1
+
UNIQUE index (track_id, cycle)
+
stores key columns + rowid again
+
+
+
Variant C · B-tree 2
+
table keyed by friction_cycle_id
+
per-row lookup to fetch cof
+
+
+
    +
  • +74% disk. In SQLite the surrogate id column is a rowid alias and costs nothing in the table itself - but the mandatory UNIQUE(track_id, cycle) index is a full second B-tree whose entries contain almost every column of the row. Two trees instead of one on a 3-column table is close to a doubling.
    • +
  • +59% on grouped aggregation (the pipeline's forced-full-scan Q5). In A/B the table is clustered by track_id, so GROUP BY track_id streams the tree in group order. In C the engine walks the UNIQUE index for ordering and hops into the table B-tree for every cof value - the classic double access path.
    • +
  • +5-13% on ranges and plain scans. More pages (190 vs 109 MiB) means more page traversals and cell decoding, even fully cached.
    • +
  • ~0% on point lookups. Tree depth is the same either way; a single-row probe does not expose the second tree.
    • +
  • PostgreSQL outlook. PG has no WITHOUT ROWID: the composite-PK table is a heap plus one PK index either way, but variant C still adds the id column to every heap tuple and the second index - expect +30-50% on the bulk tables, plus higher COPY-time FK validation cost.
    • +
+
Integrity is a constant across all variants. Every relation is a single-column integer FK. Choosing C buys naming uniformity, not more integrity - and the surrogate key it adds is never referenced by anything.
+
+
+ + +
+

Recommendation

+
+

Variant B  Composite natural PK on bulk tables + enforced foreign keys

+
    +
  • Referential integrity is guaranteed by the DBMS itself at a measured cost of +0.5 s of load time and nothing else.
    • +
  • Disk footprint and read profile are identical to the leanest variant A: 109.1 MiB, single clustered B-tree per bulk table.
    • +
  • Dimension tables (batches, wafers, coupons, runs, tracks, instruments) keep their classic single-column surrogate keys - <singular>_id, never a bare id - because other tables reference their rows.
    • +
  • Variant C remains a valid design where fact rows must be individually addressable from outside (annotations, reprocessing links). No table in this schema references an individual cycle or loop point, so its cost (+74% disk, up to +59% on aggregation, growing at 600 GB scale) buys nothing here.
    • +
+
+
+ Measured 2026-07-11 on the deterministic simulated corpus (seed 20260711, 141.8 MiB CSV) · SQLite, warm OS cache, best of 3 runs · + scripts: measure_key_variants.py (build + size + load + Q1) and read_benchmarks.py (six read patterns) · + companion page: Tribology_Ontology.html (measurement-result registry). +
+
+ +
+
+
+ + + + diff --git a/docs/rules/db-sql-schema.md b/docs/rules/db-sql-schema.md index 55278cb..1c2fd4b 100644 --- a/docs/rules/db-sql-schema.md +++ b/docs/rules/db-sql-schema.md @@ -74,11 +74,14 @@ and it wins where a spec detail is ambiguous. - Loads are all-or-nothing per table and validated against `MANIFEST.csv` row counts - see [code-error-handling.md](code-error-handling.md) section 3 and [test-pipeline-validation.md](test-pipeline-validation.md). -- On every SQLite connection that relies on FK behaviour: - `PRAGMA foreign_keys = ON` (SQLite silently ignores FK clauses without - it). Bulk tables MAY declare FK columns without enforced constraints for - load speed - row-count validation is the integrity gate; whichever choice - is made must be the same in both engines. +- **Foreign keys are ENFORCED in both engines on every table, including + bulk tables** (owner decision 2026-07-11, after measurement on the real + corpus: +0.5 s load on 4.3M rows in SQLite, zero disk and read cost; + full study with ER diagrams and measurements: + [docs/research/Tribology_FK_Architecture.html](../research/Tribology_FK_Architecture.html)). + Every SQLite connection sets `PRAGMA foreign_keys = ON` (SQLite silently + ignores FK clauses without it). Row-count validation against + `MANIFEST.csv` remains a second, independent integrity gate. ## 5. `track_summary` - precomputed, and off-limits to Q5 @@ -89,6 +92,20 @@ never read it** - Q5 is the forced full-scan benchmark over raw `friction_cycles` in every format (spec [08](../specs/08_benchmark_queries.md)). +Binding computation (identical semantics in both engines; float results +agree within the 1e-9 checksum tolerance): + +1. Steady-state proxy window = cycles 501-1000 (the second half); + `m` = avg(cof), `s` = stddev_pop(cof) over that window. +2. `run_in_cycles` = the first cycle `c` with `|cof(c) - m| < 2*s`; + fallback 500 if no cycle qualifies. +3. `cof_ss_mean` / `cof_ss_std` = avg / stddev_pop of cof over cycles + `> run_in_cycles`. + +SQLite computes this in Python during load +([common/track_summary.py](../../common/track_summary.py)); PostgreSQL as +a materialized view implementing the same three steps. + ## 6. Engine parity - Same logical schema, same table names, same column names, same row diff --git a/docs/specs/05_convert_sqlite.md b/docs/specs/05_convert_sqlite.md index 644a8fd..51c65ff 100644 --- a/docs/specs/05_convert_sqlite.md +++ b/docs/specs/05_convert_sqlite.md @@ -10,17 +10,19 @@ Build the single-file relational variant optimized for size and single-user quer CSV tree. ## Processing instructions -- One database `./out/sqlite/tribo.db`. PRAGMAs during load: `journal_mode=OFF, synchronous=OFF, cache_size=-262144`; after load: `journal_mode=WAL`, `ANALYZE`, `VACUUM`. -- Schema (16-table style, integer surrogate keys everywhere; no TEXT foreign keys in bulk tables): +- One database `./out/sqlite/tribo.db`. PRAGMAs during load: `journal_mode=OFF, synchronous=OFF, cache_size=-262144, foreign_keys=ON`; after load: `journal_mode=WAL`, `ANALYZE`, `VACUUM`. +- Foreign keys are enforced on every table, including bulk tables (variant B of the measured key-schema study; owner decision 2026-07-11, see db-sql-schema rule). Full study with ER diagrams and measurements: [../research/Tribology_FK_Architecture.html](../research/Tribology_FK_Architecture.html). +- Schema (16-table style, integer surrogate keys everywhere; no TEXT foreign keys in bulk tables; SQL column names are the lowercase of the CSV headers per data-naming-units): - `batches, wafers, coupons, runs, instruments, tracks` - dimension tables, TEXT natural keys kept as unique columns. - - `xrf_points(coupon_id INT, gx INT, gy INT, pt_wtpct REAL, au_wtpct REAL)`. - - `profilometry_points(coupon_id INT, gx INT, gy INT, thickness_um REAL)`. - - `nanoindentation(coupon_id INT, indent_id INT, x_um REAL, y_um REAL, hardness_gpa REAL, reduced_modulus_gpa REAL)`. - - `afm(coupon_id INT PRIMARY KEY, ra_nm REAL, rq_nm REAL)`. + - `simtra_profiles(batch_id INT, row_no INT, angle_deg REAL, energy_ev REAL, pt_flux REAL, au_flux REAL, PRIMARY KEY(batch_id, row_no)) WITHOUT ROWID`. + - `xrf_points(coupon_id INT, grid_x INT, grid_y INT, pt_wtpct REAL, au_wtpct REAL, PRIMARY KEY(coupon_id, grid_x, grid_y)) WITHOUT ROWID`. + - `profilometry_points(coupon_id INT, grid_x INT, grid_y INT, thickness_um REAL, PRIMARY KEY(coupon_id, grid_x, grid_y)) WITHOUT ROWID`. + - `nanoindentation(coupon_id INT, indent_id INT, x_um REAL, y_um REAL, hardness_gpa REAL, reduced_modulus_gpa REAL, max_load_mn REAL, PRIMARY KEY(coupon_id, indent_id)) WITHOUT ROWID`. + - `afm(coupon_id INT PRIMARY KEY, ra_nm REAL, rq_nm REAL, image_file TEXT)`. - `friction_cycles(track_id INT, cycle INT, cof REAL, PRIMARY KEY(track_id, cycle)) WITHOUT ROWID`. - - `friction_loop_points(track_id INT, cycle INT, pt INT, position_um REAL, force_mn REAL, PRIMARY KEY(track_id, cycle, pt)) WITHOUT ROWID`. + - `friction_loop_points(track_id INT, cycle INT, pt INT, position_um REAL, friction_force_mn REAL, PRIMARY KEY(track_id, cycle, pt)) WITHOUT ROWID`. - `wear(track_id INT PRIMARY KEY, wear_volume_um3 REAL, k_archard REAL, sliding_distance_m REAL)`. - - `track_summary(track_id INT PRIMARY KEY, cof_ss_mean REAL, cof_ss_std REAL, run_in_cycles INT)` - precomputed during load. + - `track_summary(track_id INT PRIMARY KEY, cof_ss_mean REAL, cof_ss_std REAL, run_in_cycles INT)` - precomputed during load (algorithm fixed in db-sql-schema rule, section 5). - Indexes for Q1-Q7: `coupons(au_wtpct_mean)`, `tracks(run_id)`, `tracks(coupon_id)`, `runs(environment)`, composite `tracks(environment, load_mn)` (denormalize environment/load onto tracks). - Batch inserts with `executemany`, 50k rows per transaction. - Validate row counts against MANIFEST. diff --git a/generate_data.py b/generate_data.py index 8fdb257..8713e3b 100644 --- a/generate_data.py +++ b/generate_data.py @@ -9,21 +9,26 @@ instruments would write it, driven entirely by out/config/lab_config.yaml from __future__ import annotations -import argparse import hashlib import logging import math import shutil import sys -from datetime import datetime, timedelta, timezone +from datetime import datetime, timedelta from pathlib import Path import numpy as np -import yaml + +from common.pipeline import ( + check_dependencies, + load_lab_config, + parse_task_args, + remove_stale_marker, + write_marker, +) logger = logging.getLogger(__name__) -REPO_ROOT = Path(__file__).resolve().parent TASK_ID = "03" DEPENDS_ON = ["01"] @@ -361,56 +366,17 @@ class CorpusGenerator: return len(data) -def check_dependencies(out_root: Path) -> None: - for dep in DEPENDS_ON: - marker = out_root / ".done" / f"{dep}.ok" - if not marker.exists(): - raise FileNotFoundError(f"dependency marker missing: {marker} - run task {dep} first") - - -def write_marker(marker_path: Path, gen: CorpusGenerator, tree_bytes: int) -> None: - lines = [ - f"task: '{TASK_ID}'", - "status: ok", - f"csv_root: {gen.csv_root.as_posix()}", - f"files: {len(gen.manifest)}", - f"total_rows: {gen.total_rows}", - f"total_bytes: {tree_bytes}", - f"coupons: {gen.counts['coupons']}", - f"tracks: {gen.counts['tracks']}", - f"cycle_rows: {gen.counts['cycle_rows']}", - f"loop_points: {gen.counts['loop_rows']}", - f"xrf_points: {gen.counts['xrf_rows']}", - ] - marker_path.parent.mkdir(parents=True, exist_ok=True) - with open(marker_path, "w", encoding="utf-8", newline="\n") as fh: - fh.write("\n".join(lines) + "\n") - - def main() -> int: - parser = argparse.ArgumentParser(description="Task 03: generate the CSV corpus") - parser.add_argument("--log-level", default="INFO", choices=["DEBUG", "INFO", "WARNING", "ERROR"]) - parser.add_argument("--out-root", type=Path, default=REPO_ROOT / "out", help="artifact tree root (default: ./out)") - args = parser.parse_args() - logging.basicConfig( - level=args.log_level, - stream=sys.stderr, - format="%(asctime)s %(levelname)-7s %(name)s: %(message)s", - ) - + args = parse_task_args("Task 03: generate the CSV corpus") out_root: Path = args.out_root csv_root = out_root / "csv" - marker_path = out_root / ".done" / f"{TASK_ID}.ok" try: - check_dependencies(out_root) - if marker_path.exists(): - logger.info("re-run: removing stale marker %s", marker_path) - marker_path.unlink() + check_dependencies(out_root, DEPENDS_ON) + remove_stale_marker(out_root, TASK_ID) if csv_root.exists(): logger.info("re-run: removing previous corpus %s", csv_root) shutil.rmtree(csv_root) - with open(out_root / "config" / "lab_config.yaml", encoding="utf-8") as fh: - cfg = yaml.safe_load(fh) + cfg = load_lab_config(out_root) gen = CorpusGenerator(cfg, csv_root) gen.generate() @@ -424,7 +390,17 @@ def main() -> int: raise ValidationError( f"tree size {tree_mib:.1f} MiB outside gate {SIZE_GATE_MIB[0]}-{SIZE_GATE_MIB[1]} MiB" ) - write_marker(marker_path, gen, tree_bytes) + marker_path = write_marker(out_root, TASK_ID, { + "csv_root": gen.csv_root.as_posix(), + "files": len(gen.manifest), + "total_rows": gen.total_rows, + "total_bytes": tree_bytes, + "coupons": gen.counts["coupons"], + "tracks": gen.counts["tracks"], + "cycle_rows": gen.counts["cycle_rows"], + "loop_points": gen.counts["loop_rows"], + "xrf_points": gen.counts["xrf_rows"], + }) except Exception: logger.critical("task %s failed", TASK_ID, exc_info=True) return 1 diff --git a/make_lab_config.py b/make_lab_config.py index 9805dff..1c04ec6 100644 --- a/make_lab_config.py +++ b/make_lab_config.py @@ -8,16 +8,16 @@ completion marker. Spec: docs/specs/01_lab_configuration.md. from __future__ import annotations -import argparse import logging import sys from pathlib import Path import yaml +from common.pipeline import parse_task_args, remove_stale_marker, write_marker + logger = logging.getLogger(__name__) -REPO_ROOT = Path(__file__).resolve().parent TASK_ID = "01" # Transcription of docs/specs/01_lab_configuration.md. This literal is the @@ -301,51 +301,30 @@ def write_yaml(cfg: dict, config_path: Path) -> int: return len(text.encode("utf-8")) -def write_marker(marker_path: Path, config_path: Path, config_bytes: int, cfg: dict) -> None: - v = cfg["volumes"] - f = cfg["friction_assignment"] - h = cfg["hierarchy"] - lines = [ - f"task: '{TASK_ID}'", - "status: ok", - f"config_file: {config_path.as_posix()}", - f"config_bytes: {config_bytes}", - f"seed: {cfg['seed']}", - f"batches: {h['batches']}", - f"wafers: {h['batches'] * h['wafers_per_batch']}", - f"coupons: {v['coupons_total']}", - f"friction_coupons: {f['friction_coupons_total']}", - f"tracks: {v['tracks_total']}", - f"cycle_rows: {v['cycle_rows_total']}", - f"loop_points: {v['loop_points_total']}", - f"xrf_points: {v['xrf_points_total']}", - ] - marker_path.parent.mkdir(parents=True, exist_ok=True) - with open(marker_path, "w", encoding="utf-8", newline="\n") as fh: - fh.write("\n".join(lines) + "\n") - - def main() -> int: - parser = argparse.ArgumentParser(description="Task 01: write out/config/lab_config.yaml") - parser.add_argument("--log-level", default="INFO", choices=["DEBUG", "INFO", "WARNING", "ERROR"]) - parser.add_argument("--out-root", type=Path, default=REPO_ROOT / "out", help="artifact tree root (default: ./out)") - args = parser.parse_args() - logging.basicConfig( - level=args.log_level, - stream=sys.stderr, - format="%(asctime)s %(levelname)-7s %(name)s: %(message)s", - ) - + args = parse_task_args("Task 01: write out/config/lab_config.yaml") out_root: Path = args.out_root config_path = out_root / "config" / "lab_config.yaml" - marker_path = out_root / ".done" / f"{TASK_ID}.ok" try: - if marker_path.exists(): - logger.info("re-run: removing stale marker %s", marker_path) - marker_path.unlink() + remove_stale_marker(out_root, TASK_ID) validate_config(LAB_CONFIG) config_bytes = write_yaml(LAB_CONFIG, config_path) - write_marker(marker_path, config_path, config_bytes, LAB_CONFIG) + v = LAB_CONFIG["volumes"] + f = LAB_CONFIG["friction_assignment"] + h = LAB_CONFIG["hierarchy"] + marker_path = write_marker(out_root, TASK_ID, { + "config_file": config_path.as_posix(), + "config_bytes": config_bytes, + "seed": LAB_CONFIG["seed"], + "batches": h["batches"], + "wafers": h["batches"] * h["wafers_per_batch"], + "coupons": v["coupons_total"], + "friction_coupons": f["friction_coupons_total"], + "tracks": v["tracks_total"], + "cycle_rows": v["cycle_rows_total"], + "loop_points": v["loop_points_total"], + "xrf_points": v["xrf_points_total"], + }) except Exception: logger.critical("task %s failed", TASK_ID, exc_info=True) return 1 diff --git a/make_process_flow.py b/make_process_flow.py index 15b4bc5..9b1a466 100644 --- a/make_process_flow.py +++ b/make_process_flow.py @@ -8,16 +8,20 @@ the laboratory workflow: ./out/report/process_flow.md plus one standalone from __future__ import annotations -import argparse import logging import sys from pathlib import Path -import yaml +from common.pipeline import ( + check_dependencies, + load_lab_config, + parse_task_args, + remove_stale_marker, + write_marker, +) logger = logging.getLogger(__name__) -REPO_ROOT = Path(__file__).resolve().parent TASK_ID = "02" DEPENDS_ON = ["01"] @@ -190,13 +194,6 @@ def build_diagrams(cfg: dict) -> list[tuple[str, str, str, str]]: ] -def check_dependencies(out_root: Path) -> None: - for dep in DEPENDS_ON: - marker = out_root / ".done" / f"{dep}.ok" - if not marker.exists(): - raise FileNotFoundError(f"dependency marker missing: {marker} - run task {dep} first") - - def validate_diagram(stem: str, text: str) -> None: first = text.splitlines()[0].strip() if text.strip() else "" if not first.startswith(DIAGRAM_TYPES): @@ -235,42 +232,20 @@ def write_outputs(cfg: dict, out_root: Path) -> tuple[Path, int, int]: return md_path, len(md_text.encode("utf-8")), total_diagram_bytes -def write_marker(marker_path: Path, md_path: Path, md_bytes: int, diagram_bytes: int) -> None: - lines = [ - f"task: '{TASK_ID}'", - "status: ok", - f"process_flow_file: {md_path.as_posix()}", - f"process_flow_bytes: {md_bytes}", - "diagrams: 6", - f"diagram_bytes: {diagram_bytes}", - ] - marker_path.parent.mkdir(parents=True, exist_ok=True) - with open(marker_path, "w", encoding="utf-8", newline="\n") as fh: - fh.write("\n".join(lines) + "\n") - - def main() -> int: - parser = argparse.ArgumentParser(description="Task 02: write process flow diagrams") - parser.add_argument("--log-level", default="INFO", choices=["DEBUG", "INFO", "WARNING", "ERROR"]) - parser.add_argument("--out-root", type=Path, default=REPO_ROOT / "out", help="artifact tree root (default: ./out)") - args = parser.parse_args() - logging.basicConfig( - level=args.log_level, - stream=sys.stderr, - format="%(asctime)s %(levelname)-7s %(name)s: %(message)s", - ) - + args = parse_task_args("Task 02: write process flow diagrams") out_root: Path = args.out_root - marker_path = out_root / ".done" / f"{TASK_ID}.ok" try: - check_dependencies(out_root) - if marker_path.exists(): - logger.info("re-run: removing stale marker %s", marker_path) - marker_path.unlink() - with open(out_root / "config" / "lab_config.yaml", encoding="utf-8") as fh: - cfg = yaml.safe_load(fh) + check_dependencies(out_root, DEPENDS_ON) + remove_stale_marker(out_root, TASK_ID) + cfg = load_lab_config(out_root) md_path, md_bytes, diagram_bytes = write_outputs(cfg, out_root) - write_marker(marker_path, md_path, md_bytes, diagram_bytes) + marker_path = write_marker(out_root, TASK_ID, { + "process_flow_file": md_path.as_posix(), + "process_flow_bytes": md_bytes, + "diagrams": 6, + "diagram_bytes": diagram_bytes, + }) except Exception: logger.critical("task %s failed", TASK_ID, exc_info=True) return 1