From d8e8490e050778d0bed6f5c970996f58550f68e7 Mon Sep 17 00:00:00 2001 From: Dave Rogers Date: Fri, 10 Jul 2026 23:35:42 +1000 Subject: [PATCH] Validate AnVIL metadata records at load (#161) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Nothing between the AnVIL API and the classifiers checked the shape of a record. _load_input validated only the envelope. A renamed key or a newly-nullable column flowed into 758,658 records and surfaced as a crash deep in a multi-hour run — or as a silently wrong classification. Two contracts, deliberately separate: - ANVIL_RECORD_SCHEMA: the full 12-field record the download produces. Checked by `make validate-metadata` before a run. - validate_pipeline_records: the far smaller promise ClassifyPipeline relies on. A derived input (a filtered subset, a test fixture) carries only a few fields, so imposing the AnVIL contract there would be wrong. Two constraints are measured, not assumed, and a naive validator gets both wrong: * file_size may be 0 — three corpus records are. min_value=0, not 1. * data_modality and reference_assembly are AnVIL's own declarations and are null for ~99% of files. Nullable by design. No coercion and no silent drops. A None is never turned into "": that hides the drift this exists to catch, and is exactly what makes `rec.get(k, "")` dangerous — the default fires only when the key is absent, so a present-but-null value returns None and crashes the first `.lower()` downstream. A dropped record is indistinguishable from a file that was never seen (#155), so _load_input raises. Unknown fields are not a problem: the API may add a column. A renamed one shows up as `missing:`. Verified against the real corpus: 758,658 records, no problems, 4.0s. Injecting a renamed key, a null, a stringified size and an uppercase md5 produces all four problem kinds with bounded samples. validate_pipeline_records adds 0.48s per run. Co-Authored-By: Claude Opus 4.8 --- Makefile | 6 +- scripts/validate_metadata.py | 76 +++++++++++++ src/meta_disco/metadata_schema.py | 173 ++++++++++++++++++++++++++++++ src/meta_disco/pipeline.py | 42 +++++--- tests/test_metadata_schema.py | 156 +++++++++++++++++++++++++++ tests/test_pipeline.py | 4 +- 6 files changed, 441 insertions(+), 16 deletions(-) create mode 100644 scripts/validate_metadata.py create mode 100644 src/meta_disco/metadata_schema.py create mode 100644 tests/test_metadata_schema.py diff --git a/Makefile b/Makefile index c8f7760..90b879d 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: test lint classify classify-hprc classify-and-report download classify-bam classify-vcf classify-fastq classify-fasta classify-headers classify-bed coverage-report validation-report all-reports download-hprc validate-hprc clean help +.PHONY: test lint validate-metadata classify classify-hprc classify-and-report download classify-bam classify-vcf classify-fastq classify-fasta classify-headers classify-bed coverage-report validation-report all-reports download-hprc validate-hprc clean help help: @echo "meta-disco — AnVIL file metadata classification" @@ -8,6 +8,7 @@ help: @echo " make classify Run full classification pipeline (all file types, parallel)" @echo " make classify-and-report Run classify + regenerate all reports" @echo " make download Download fresh AnVIL metadata from API" + @echo " make validate-metadata Check downloaded metadata against the record schema" @echo "" @echo " make classify-bam Classify BAM/CRAM files (network required)" @echo " make classify-vcf Classify VCF files (network required)" @@ -41,6 +42,9 @@ classify-and-report: classify classify-hprc all-reports download: python scripts/download_anvil_metadata.py +validate-metadata: + python scripts/validate_metadata.py + classify-headers: classify-bam classify-vcf classify-fastq classify-fasta classify-bam: diff --git a/scripts/validate_metadata.py b/scripts/validate_metadata.py new file mode 100644 index 0000000..bf6a181 --- /dev/null +++ b/scripts/validate_metadata.py @@ -0,0 +1,76 @@ +#!/usr/bin/env python3 +"""Validate downloaded AnVIL metadata against the record schema. + +Run this after `make download` and before a multi-hour classification run: a +renamed key or a newly-nullable column otherwise surfaces as a crash somewhere +deep in the corpus, or as a silently wrong classification. + + python scripts/validate_metadata.py + python scripts/validate_metadata.py --input data/anvil/anvil_files_metadata.json + +Exit code 1 when the records do not match the schema. +""" + +import argparse +import json +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from src.meta_disco.metadata_schema import ( # noqa: E402 + ANVIL_RECORD_SCHEMA, + validate_anvil_records, +) + +DEFAULT_INPUT = Path("data/anvil/anvil_files_metadata.json") + + +def load_records(path: Path) -> list[dict]: + """Read the records array from a metadata file (JSON or NDJSON).""" + if path.suffix == ".ndjson": + with open(path) as f: + return [json.loads(line) for line in f if line.strip()] + + with open(path) as f: + data = json.load(f) + if not isinstance(data, dict): + raise TypeError(f"Expected a JSON object, got {type(data).__name__}") + records = data.get("files") or data.get("results") + if records is None: + raise ValueError("JSON object must contain a 'files' or 'results' key") + return records + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--input", type=Path, default=DEFAULT_INPUT, + help=f"Metadata file to check (default: {DEFAULT_INPUT})") + args = parser.parse_args() + + if not args.input.is_file(): + print(f"No such file: {args.input}", file=sys.stderr) + return 1 + + records = load_records(args.input) + report = validate_anvil_records(records) + + print(f"Schema: {len(ANVIL_RECORD_SCHEMA)} fields") + print(f"Input : {args.input}") + print(report.summary()) + + if report.is_valid: + return 0 + + print( + "\nThe AnVIL record shape has changed, or the download is incomplete.\n" + "Fix the schema in src/meta_disco/metadata_schema.py if the change is\n" + "intended — do NOT coerce the values, or the drift is hidden rather than\n" + "caught.", + file=sys.stderr, + ) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/meta_disco/metadata_schema.py b/src/meta_disco/metadata_schema.py new file mode 100644 index 0000000..7e3babf --- /dev/null +++ b/src/meta_disco/metadata_schema.py @@ -0,0 +1,173 @@ +"""Shape validation for AnVIL file-metadata records. + +Nothing between the AnVIL API and the classifiers checks the shape of a record. +A renamed key, a newly-nullable column, or a stringified size flows silently into +758k records and surfaces as a crash — or a wrong classification — somewhere deep +in a multi-hour run. + +Two contracts, deliberately separate: + +* :data:`ANVIL_RECORD_SCHEMA` — the full 12-field record the download script + produces. Checked by ``scripts/validate_metadata.py`` and at download time. +* :func:`validate_pipeline_records` — the far smaller promise ``ClassifyPipeline`` + relies on. Derived inputs (a filtered subset, a hand-built test fixture) carry + only a few fields, so requiring the full AnVIL record here would be wrong. + +Nothing here mutates a record. A ``None`` is never coerced to ``""``: that would +hide the drift this exists to catch, and it is what makes ``rec.get(k, "")`` +dangerous — the default fires only when the key is *absent*, so a present-but-null +value returns ``None`` and crashes the first ``.lower()`` downstream. +""" + +import re +from collections import Counter +from dataclasses import dataclass, field + +MD5_PATTERN = re.compile(r"^[0-9a-f]{32}$") + +# Bounded so a systematically broken download reports a summary, not 758k lines. +MAX_SAMPLES_PER_PROBLEM = 3 + + +@dataclass(frozen=True) +class FieldSpec: + """The contract for one record field.""" + + types: tuple[type, ...] + nullable: bool = False + allow_empty: bool = False + pattern: re.Pattern | None = None + min_value: int | None = None + + +# Measured against the 758,658-record corpus. Two constraints are easy to get +# wrong and are called out rather than inferred: +# * file_size may be 0 — three records are. `min_value=0`, not 1. +# * data_modality and reference_assembly are AnVIL's own declarations and are +# null for ~99% of files. They are nullable by design, not by accident. +ANVIL_RECORD_SCHEMA: dict[str, FieldSpec] = { + "entry_id": FieldSpec((str,)), + "file_id": FieldSpec((str,)), + "file_name": FieldSpec((str,)), + "file_format": FieldSpec((str,)), + "file_size": FieldSpec((int,), min_value=0), + "file_md5sum": FieldSpec((str,), pattern=MD5_PATTERN), + "drs_uri": FieldSpec((str,)), + "dataset_id": FieldSpec((str,)), + "dataset_title": FieldSpec((str,)), + "is_supplementary": FieldSpec((bool,)), + "data_modality": FieldSpec((str,), nullable=True), + "reference_assembly": FieldSpec((str,), nullable=True), +} + +# Fields ClassifyPipeline reads. Each may be absent (a derived input need not +# carry it) but must never be present-and-null, which is the failure this guards. +PIPELINE_STRING_FIELDS = ("file_md5sum", "file_name", "file_format", "dataset_title") + + +@dataclass +class ValidationReport: + """Problems found, counted by kind, with a bounded sample of each.""" + + total: int = 0 + counts: Counter = field(default_factory=Counter) + samples: dict[str, list[str]] = field(default_factory=dict) + + @property + def is_valid(self) -> bool: + return not self.counts + + @property + def problem_records(self) -> int: + """Number of distinct problems recorded, not of distinct records. + + One record may contribute several problems, so this is an upper bound on + the number of bad records — it is a signal of scale, not a count. + """ + return sum(self.counts.values()) + + def add(self, kind: str, detail: str) -> None: + self.counts[kind] += 1 + bucket = self.samples.setdefault(kind, []) + if len(bucket) < MAX_SAMPLES_PER_PROBLEM: + bucket.append(detail) + + def summary(self) -> str: + if self.is_valid: + return f"{self.total:,} records, no problems." + lines = [f"{self.total:,} records, {len(self.counts)} kinds of problem:"] + for kind, n in self.counts.most_common(): + lines.append(f" {kind}: {n:,}") + for detail in self.samples.get(kind, []): + lines.append(f" e.g. {detail}") + return "\n".join(lines) + + +def _check_value(name: str, value, spec: FieldSpec) -> str | None: + """The problem this value has against its spec, or None.""" + if value is None: + return None if spec.nullable else "null" + + # bool is a subclass of int, so an `is_supplementary` in a `file_size` slot + # would silently pass an isinstance(int) check. + if isinstance(value, bool) and bool not in spec.types: + return "type" + if not isinstance(value, spec.types): + return "type" + + if isinstance(value, str): + if not value and not spec.allow_empty: + return "empty" + if spec.pattern and not spec.pattern.match(value): + return "pattern" + if spec.min_value is not None and value < spec.min_value: + return "range" + return None + + +def validate_anvil_records( + records: list[dict], schema: dict[str, FieldSpec] = ANVIL_RECORD_SCHEMA +) -> ValidationReport: + """Validate records against the full AnVIL contract. + + Unknown fields are *not* a problem: the API may add a column, and that should + not fail a download. A renamed column shows up as `missing:`. + """ + report = ValidationReport(total=len(records)) + for i, rec in enumerate(records): + if not isinstance(rec, dict): + report.add("not_an_object", f"record {i}: {type(rec).__name__}") + continue + for name, spec in schema.items(): + if name not in rec: + report.add(f"missing:{name}", f"record {i}") + continue + problem = _check_value(name, rec[name], spec) + if problem: + report.add(f"{problem}:{name}", f"record {i}: {rec[name]!r}") + return report + + +def validate_pipeline_records(records: list[dict]) -> ValidationReport: + """Validate the small promise ClassifyPipeline relies on. + + Each of PIPELINE_STRING_FIELDS may be absent, but must not be present-and-null: + `rec.get("file_name", "")` returns None for a null, and the first `.lower()` or + slice downstream then raises. `file_size`, when present, must be a non-negative + int (or null — some derived inputs omit the size rather than the key). + """ + report = ValidationReport(total=len(records)) + for i, rec in enumerate(records): + if not isinstance(rec, dict): + report.add("not_an_object", f"record {i}: {type(rec).__name__}") + continue + for name in PIPELINE_STRING_FIELDS: + if name in rec and not isinstance(rec[name], str): + kind = "null" if rec[name] is None else "type" + report.add(f"{kind}:{name}", f"record {i}: {rec[name]!r}") + size = rec.get("file_size") + if size is not None and (isinstance(size, bool) or not isinstance(size, int)): + report.add("type:file_size", f"record {i}: {size!r}") + elif isinstance(size, int) and not isinstance(size, bool) and size < 0: + report.add("range:file_size", f"record {i}: {size!r}") + return report diff --git a/src/meta_disco/pipeline.py b/src/meta_disco/pipeline.py index 70d82f2..1615eee 100644 --- a/src/meta_disco/pipeline.py +++ b/src/meta_disco/pipeline.py @@ -12,6 +12,8 @@ from threading import Lock from typing import Callable +from .metadata_schema import validate_pipeline_records + @dataclass(frozen=True) class FileTypeConfig: @@ -153,21 +155,35 @@ def classify_single( # --- Internal --- def _load_input(self) -> list[dict]: - """Load NDJSON or JSON input, extracting records array.""" + """Load NDJSON or JSON input, extract the records array, and validate it. + + Validates only the fields this pipeline reads (``validate_pipeline_records``), + not the full AnVIL record: a derived input may legitimately carry a subset. + Raises rather than dropping the offenders — a dropped record is + indistinguishable from a file that was never seen (see #155). + """ with open(self.input_path) as f: if self.input_path.suffix == ".ndjson": - return [json.loads(line) for line in f if line.strip()] - data = json.load(f) - if not isinstance(data, dict): - raise TypeError( - f"Expected JSON object with 'results' key, got {type(data).__name__}" - ) - records = data.get("results") or data.get("files") - if records is None: - raise ValueError( - "JSON object must contain a 'results' or 'files' key" - ) - return records + records = [json.loads(line) for line in f if line.strip()] + else: + data = json.load(f) + if not isinstance(data, dict): + raise TypeError( + f"Expected JSON object with 'results' key, got {type(data).__name__}" + ) + records = data.get("results") or data.get("files") + if records is None: + raise ValueError( + "JSON object must contain a 'results' or 'files' key" + ) + + report = validate_pipeline_records(records) + if not report.is_valid: + raise ValueError( + f"{self.input_path} holds records this pipeline cannot read.\n" + f"{report.summary()}" + ) + return records def _filter_records(self, records: list[dict]) -> list[dict]: """Filter to records matching file_type.extensions with valid MD5.""" diff --git a/tests/test_metadata_schema.py b/tests/test_metadata_schema.py new file mode 100644 index 0000000..9540014 --- /dev/null +++ b/tests/test_metadata_schema.py @@ -0,0 +1,156 @@ +"""Tests for AnVIL metadata record validation. + +The constraints here were measured against the 758,658-record corpus, not +assumed. Two of them are counterintuitive and are pinned explicitly: `file_size` +may be 0, and `data_modality`/`reference_assembly` are nullable by design. +""" + +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from src.meta_disco.metadata_schema import ( + ANVIL_RECORD_SCHEMA, + validate_anvil_records, + validate_pipeline_records, +) + + +def _record(**overrides): + """A record that satisfies the full AnVIL schema.""" + rec = { + "entry_id": "e1", + "file_id": "f1", + "file_name": "sample.bam", + "file_format": ".bam", + "file_size": 1234, + "file_md5sum": "0" * 32, + "drs_uri": "drs://drs.anv0:v2_abc", + "dataset_id": "d1", + "dataset_title": "ANVIL_HPRC", + "is_supplementary": False, + "data_modality": None, + "reference_assembly": None, + } + rec.update(overrides) + return rec + + +class TestAnvilSchema: + def test_a_good_record_has_no_problems(self): + assert validate_anvil_records([_record()]).is_valid + + def test_zero_file_size_is_valid(self): + """Three corpus records have size 0. `min_value=0`, not 1.""" + assert validate_anvil_records([_record(file_size=0)]).is_valid + + def test_negative_file_size_is_a_range_problem(self): + report = validate_anvil_records([_record(file_size=-1)]) + assert "range:file_size" in report.counts + + def test_declared_modality_and_assembly_are_nullable(self): + """AnVIL's own declarations, null for ~99% of the corpus.""" + report = validate_anvil_records( + [_record(data_modality=None, reference_assembly=None)] + ) + assert report.is_valid + + def test_a_null_required_field_is_a_null_problem(self): + report = validate_anvil_records([_record(file_name=None)]) + assert report.counts["null:file_name"] == 1 + + def test_a_renamed_key_shows_up_as_missing(self): + rec = _record() + rec["fileName"] = rec.pop("file_name") + report = validate_anvil_records([rec]) + assert report.counts["missing:file_name"] == 1 + + def test_an_added_key_is_not_a_problem(self): + """The API may add a column; that must not fail a download.""" + assert validate_anvil_records([_record(new_column="x")]).is_valid + + def test_a_stringified_size_is_a_type_problem(self): + report = validate_anvil_records([_record(file_size="1234")]) + assert report.counts["type:file_size"] == 1 + + def test_a_bool_is_not_an_int(self): + """bool subclasses int, so isinstance(True, int) is True.""" + report = validate_anvil_records([_record(file_size=True)]) + assert report.counts["type:file_size"] == 1 + + def test_an_empty_string_is_an_empty_problem(self): + report = validate_anvil_records([_record(dataset_title="")]) + assert report.counts["empty:dataset_title"] == 1 + + def test_a_malformed_md5_is_a_pattern_problem(self): + report = validate_anvil_records([_record(file_md5sum="NOTHEX")]) + assert report.counts["pattern:file_md5sum"] == 1 + + def test_uppercase_md5_is_rejected(self): + report = validate_anvil_records([_record(file_md5sum="A" * 32)]) + assert report.counts["pattern:file_md5sum"] == 1 + + def test_a_non_object_record_is_reported_not_crashed_on(self): + report = validate_anvil_records([_record(), "not a dict", 42]) + assert report.counts["not_an_object"] == 2 + + def test_samples_are_bounded(self): + report = validate_anvil_records([_record(file_name=None) for _ in range(50)]) + assert report.counts["null:file_name"] == 50 + assert len(report.samples["null:file_name"]) == 3 + + def test_summary_names_every_problem_kind(self): + report = validate_anvil_records([_record(file_name=None, file_size=-1)]) + summary = report.summary() + assert "null:file_name" in summary + assert "range:file_size" in summary + + def test_schema_covers_the_twelve_fields_the_download_produces(self): + assert len(ANVIL_RECORD_SCHEMA) == 12 + + +class TestPipelineContract: + """The pipeline reads four string fields and a size; a derived input may omit + any of them, but must never carry a present-and-null value.""" + + def test_a_minimal_record_is_valid(self): + assert validate_pipeline_records([{"file_md5sum": "x"}]).is_valid + + def test_absent_fields_are_fine(self): + assert validate_pipeline_records([{}]).is_valid + + def test_a_null_file_name_is_rejected(self): + """`rec.get("file_name", "")` returns None for a null, and the first + `.lower()` or slice downstream then raises.""" + report = validate_pipeline_records([{"file_md5sum": "x", "file_name": None}]) + assert report.counts["null:file_name"] == 1 + + @pytest.mark.parametrize( + "fld", ["file_md5sum", "file_name", "file_format", "dataset_title"] + ) + def test_every_pipeline_string_field_rejects_null(self, fld): + report = validate_pipeline_records([{fld: None}]) + assert report.counts[f"null:{fld}"] == 1 + + def test_a_null_file_size_is_allowed(self): + """Some derived inputs omit the size rather than the key.""" + assert validate_pipeline_records([{"file_size": None}]).is_valid + + def test_a_stringified_size_is_rejected(self): + report = validate_pipeline_records([{"file_size": "100"}]) + assert report.counts["type:file_size"] == 1 + + def test_a_bool_size_is_rejected(self): + report = validate_pipeline_records([{"file_size": True}]) + assert report.counts["type:file_size"] == 1 + + def test_zero_size_is_allowed_negative_is_not(self): + assert validate_pipeline_records([{"file_size": 0}]).is_valid + assert not validate_pipeline_records([{"file_size": -1}]).is_valid + + def test_the_pipeline_contract_does_not_require_the_anvil_fields(self): + """drs_uri, entry_id, etc. are absent from a derived input and that is fine.""" + assert validate_pipeline_records([{"file_md5sum": "x", "file_name": "a.bam"}]).is_valid diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 40d4e63..6acf9a9 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -1,11 +1,11 @@ """Tests for the shared ClassifyPipeline infrastructure.""" import json -import pytest from pathlib import Path -from src.meta_disco.pipeline import ClassifyPipeline, FileTypeConfig, NdjsonWriter +import pytest +from src.meta_disco.pipeline import ClassifyPipeline, FileTypeConfig, NdjsonWriter # --- Test fixtures ---