diff --git a/CLAUDE.md b/CLAUDE.md index 3b7e164f..7cfcc851 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -136,7 +136,7 @@ The frontend (`apps/web/`) is a Next.js 16 app with Tailwind v4. API URL is conf - `export_evaluation_targets` (`ExportEvaluationTargetsOperation`): writes an `EvaluationSet`'s delta proteins as a FASTA and publishes it to the artifact store, so the same bytes are both the LAFA `--query_file` and the `QuerySet` the internal grid dispatches against. Exists to close an asymmetry: `apps/method_runtime/protea_predict.py` REQUIRES `--query_file`, while `PredictGOTermsPayload` leaves `query_set_id` and `query_accessions` optional and silently selects every protein with an embedding when neither is given. That is a comparability defect before it is a cost one, since a grid scoring one population and a submission scoring another produce numbers that are not about the same thing. Targets are the union of the `nk` / `lk` / `pk` maps, the proteins that GAINED annotation; `removed` holds proteins that only lost annotation, which the evaluation reports and never scores, and requesting it as a category raises. Accessions are sorted and sequences wrapped at a fixed width so the same evaluation set always produces the same bytes, and the returned sha256 is what lets both consumers prove they scored one population rather than assert it. Headers are bare accessions, which round-trip through LAFA's parser unchanged. Delta proteins with no stored sequence are emitted as a warning rather than dropped silently. Runs on `protea.jobs`. - `declare_query_set_provenance` (`DeclareQuerySetProvenanceOperation`): recovers a hand-registered `QuerySet`'s derivation by **verifying** it rather than reading it out of the description. Four sets were registered through the FASTA upload endpoint by a person, with the window as an id prefix, the job as a word and the sha as a string nothing compared against; migration `c5e8d1a30b72` gave the table the columns and deliberately declared nothing, because recovering a derivation is a judgement that belongs in a job with its events. For each named set it finds every SUCCEEDED `export_evaluation_targets` whose published target count equals the entry count, fetches each one's FASTA through `fetch_and_verify` with the key **derived** from what the job declared (`targets_key_for`, the one spelling, so no later reader has to split a stored uri the way the ground truth's second door did), and writes the window, sha256, job and revision only when the accession set in those bytes equals the set's entries exactly. The uri it stores is the one that export's own `put` returned, taken from the event and checked to END in the derived key: neither is parsed, and building one here would mean guessing the bucket. The revision is not picked either, since a requeued job has more than one start: it counts the DISTINCT values and leaves NULL unless there is exactly one, because for a provenance field "cannot say" beats "the earlier one". Measured 2026-11-08: no export job records that field in any event, so all four recoveries get NULL and the operation reports it rather than writing a quiet one. Verification happens BEFORE the collapse, which the data forces: two jobs published the same digest for the same window under different `output_name` values, so the same bytes live under two keys, and collapsing first could keep the one whose object is gone. A count that matches is a coincidence waiting to happen; a set that matches is the population. Jobs publishing the same `(window, digest)` are one export run twice and collapse to the earliest; two publishing different ones are an ambiguity it refuses rather than resolves. A set with no candidate stays `legacy_undeclared` and is reported, which is the honest outcome for the one from the quick-annotation flow. A description somebody wrote is kept. `dry_run` writes nothing. Runs on `protea.jobs`. -- `export_transitions_dataset` (`ExportTransitionsDatasetOperation`): writes the ADR-D56 transitions dataset's declared tables for named windows on a declared term axis. Separate from `export_research_dataset`, which keeps writing the D45 pair table: two datasets, and mixing them would make one `dataset` row unable to say which it is, so the names here carry a `transitions-` prefix. **Windows are named by id, never by a version pair**, because the pair stopped identifying a window (62 of 69 pairs carry more than one), and each is refused if it pivots on another axis than the one declared or was built by a revision other than the running one unless the payload names that revision; admitting them in bulk with a boolean would be the same as not checking. `cut` (the `ontology_snapshot_id` indexing `terms` and `term_ancestors`) and `cut_version` (the GOA release the laboratory's leakage rule compares against) are two fields and neither is derived from the other. It refuses a window whose `version_t_prime` is after `cut_version`, because its gain is in the model's future: the first run declared `cut_version` 211 over 64 windows and **15 of them ended after 211**, all inside the span the long validation window scores. Each window's truth, releases and horizon are read ONCE into a `WindowFacts`, since two loads are two chances for one bundle's tables to be about different reads, and a window missing a `source_version` or a publication date is refused rather than written as NULL (a NULL version does not fail the leakage rule, it skips it). Today it writes five of the ten: `window_terms_out`, the axis terms a window's rule leaves out; `residue_refs`, which names where each sequence's per-residue codes live (683,887 references over 5,344 shards, not one byte of code copied) and covers the WHOLE bank, a superset of any window selection, refusing a config with no rows because the rows are keyed by the VOCABULARY config and not by the model config it was derived from, which has zero; `terms` and `term_ancestors`, which are keyed by `cut` ALONE and so need no window list at all (43,558 live terms with their aspect and 595,072 closure pairs on the campaign axis, measured, against the 783,477 `Dag.closure`'s docstring quotes for a different release); and `queries`, whose population is `delta_accessions` imported from the target export rather than restated; both carry the full `evaluation_set_id` as the join key and the writer asserts the two tables agree about their window set before publishing either, because the loader joins them as text and a mismatch silently stops restricting the negatives. It **verifies the vectors before writing them: `terms` carries `vector_uri` and `vector_sha256` for the axis's text-embedding artefact, and the operation FETCHES those bytes and checks the digest before publishing either column, refusing when the artefact is absent, because a sha copied from a message is a locator for bytes nobody checked. It writes no `Dataset` row** while any of the ten tables is unwritten, since a half bundle with a sha is a bundle that lies. Asking for a table ADR-D56 declares but this revision does not write yet is refused rather than skipped. Runs on `protea.training`. +- `export_transitions_dataset` (`ExportTransitionsDatasetOperation`): writes the ADR-D56 transitions dataset's declared tables for named windows on a declared term axis. Separate from `export_research_dataset`, which keeps writing the D45 pair table: two datasets, and mixing them would make one `dataset` row unable to say which it is, so the names here carry a `transitions-` prefix. **Windows are named by id, never by a version pair**, because the pair stopped identifying a window (62 of 69 pairs carry more than one), and each is refused if it pivots on another axis than the one declared or was built by a revision other than the running one unless the payload names that revision; admitting them in bulk with a boolean would be the same as not checking. `cut` (the `ontology_snapshot_id` indexing `terms` and `term_ancestors`) and `cut_version` (the GOA release the laboratory's leakage rule compares against) are two fields and neither is derived from the other. It refuses a window whose `version_t_prime` is after `cut_version`, because its gain is in the model's future: the first run declared `cut_version` 211 over 64 windows and **15 of them ended after 211**, all inside the span the long validation window scores. Each window's truth, releases and horizon are read ONCE into a `WindowFacts`, since two loads are two chances for one bundle's tables to be about different reads, and a window missing a `source_version` or a publication date is refused rather than written as NULL (a NULL version does not fail the leakage rule, it skips it). Today it writes six of the ten, including `gains` (the truth itself: which terms each query gained, with its stratum and the closure the gain is expressed in, 9.6M rows over the 65 admissible windows, sharing the query population as a guard rather than a filter so a gain for a protein `queries` excludes is refused rather than dropped); `window_terms_out`, the axis terms a window's rule leaves out; `residue_refs`, which names where each sequence's per-residue codes live (683,887 references over 5,344 shards, not one byte of code copied) and covers the WHOLE bank, a superset of any window selection, refusing a config with no rows because the rows are keyed by the VOCABULARY config and not by the model config it was derived from, which has zero; `terms` and `term_ancestors`, which are keyed by `cut` ALONE and so need no window list at all (43,558 live terms with their aspect and 595,072 closure pairs on the campaign axis, measured, against the 783,477 `Dag.closure`'s docstring quotes for a different release); and `queries`, whose population is `delta_accessions` imported from the target export rather than restated; both carry the full `evaluation_set_id` as the join key and the writer asserts the two tables agree about their window set before publishing either, because the loader joins them as text and a mismatch silently stops restricting the negatives. It **verifies the vectors before writing them: `terms` carries `vector_uri` and `vector_sha256` for the axis's text-embedding artefact, and the operation FETCHES those bytes and checks the digest before publishing either column, refusing when the artefact is absent, because a sha copied from a message is a locator for bytes nobody checked. It writes no `Dataset` row** while any of the ten tables is unwritten, since a half bundle with a sha is a bundle that lies. Asking for a table ADR-D56 declares but this revision does not write yet is refused rather than skipped. Runs on `protea.training`. These three operations form the **post-reranker functional-enrichment stage, the last component to be developed before uploading results to LAFA**. All three are fully wired and registered in the `OperationRegistry`; the remaining prerequisite is an InterProScan binary install on the host. diff --git a/docs/source/adr/D56-transitions-dataset-is-tables-keyed-by-what-they-depend-on.rst b/docs/source/adr/D56-transitions-dataset-is-tables-keyed-by-what-they-depend-on.rst index 7cba2f5f..3a69f83c 100644 --- a/docs/source/adr/D56-transitions-dataset-is-tables-keyed-by-what-they-depend-on.rst +++ b/docs/source/adr/D56-transitions-dataset-is-tables-keyed-by-what-they-depend-on.rst @@ -330,6 +330,40 @@ constraint this section states is only the bound: for a bundle that must not leak into the long validation window, ``cut_version`` has to be at or before the release that window opens on. +``gains`` shares the query population, and that is a guard +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +Noted 2026-11-08, when the table was written. ``queries`` excludes a protein +with no stored sequence, because the exported FASTA excludes it and the +training rows have to be about the scored population. So a gain for a protein +outside that set would be a gain for a query the loader does not know, and the +two tables would disagree about who the queries are **without either being +wrong on its own**. + +The writer therefore computes the query population once and refuses a gain +outside it, rather than dropping one. Dropping would make ``gains`` quietly +smaller than the truth it copies, and a row count is the only thing a reader +could compare it against. Measured over the 65 admissible windows: **zero** +delta proteins lack a sequence, so the guard changes nothing today and says so +when that stops being true. The same shape covers a gained term the axis cannot +name: the loader groups by aspect, so such a row is one it cannot place. + +**The closure is derived from the truth profile in one place.** ``projected`` +means the gain is expressed on the pivot and ``as_closed`` on each end's own +graph, and the same window under two closures is two different truths. A +profile with no declared closure is refused rather than guessed, and a test +asserts that every value of ``TRUTH_PROFILES`` has one, so adding a profile +forces someone to say which graph it closes in at the moment it is declared +rather than at run time. + +**The shape, measured before the code was written.** Over those 65 windows +``gains`` is **9.596.563 rows**, the largest window 691.553 and the mean +147.639. Six short columns at that count is the one place in this dataset where +building a tuple per row would hurt, so the columns are six parallel lists +appended to per window, and the order is stable by construction rather than by +a final sort: windows in order, strata in their declared order, queries sorted, +terms sorted. + The window is the join key, and it is the full id ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ @@ -372,6 +406,21 @@ describes and nothing would see it. Measured over the 131 evaluation sets in the database, two values are enough and a third is not needed: **122 are ``native`` and 9 are ``pivot``**. +**AND THE WRITER MADE EXACTLY THAT MISTAKE, twice, on 2026-11-08.** The first +version of ``gains`` mapped ``projected`` to ``pivot`` and ``as_closed`` to +``native``, in a constant named ``_CLOSURE_OF_PROFILE``, with a test that +walked the values of ``TRUTH_PROFILES`` and therefore could not see it. The +laboratory had made the same mapping a day earlier and withdrawn it. Both times +the paragraph below was already in this record, written by an earlier PR of +this same project: the warning was not missing, it was not read. + +So the derivation now compares the two natives against the pivot and against +each end's corpus ontology, the way the window inventory already does, a third +combination is refused rather than named, and the test walks combinations of +NATIVES. The case that settles it is in the database: the twins of +``160 -> 177`` are both ``projected`` and differ only here, so under the +mapping the (B) one would have been written as ``pivot``. + **It is NOT ``truth_profile``.** That field says how the truth is derived, ``projected`` or ``as_closed``; ``closure`` says which graphs the two ends close in. They vary independently, and the probe that settles the open decision diff --git a/protea/core/operations/export_transitions_dataset.py b/protea/core/operations/export_transitions_dataset.py index 46424e30..9dd49a91 100644 --- a/protea/core/operations/export_transitions_dataset.py +++ b/protea/core/operations/export_transitions_dataset.py @@ -47,6 +47,8 @@ ) from protea.core.operations.export_evaluation_targets import SCORED_CATEGORIES from protea.core.training_dump._window_choice import _refuse_wrong_pivot, window_identity +from protea.core.transitions_export._gains import WindowGains, gains_of, new_columns +from protea.core.transitions_export._gains import write_table as write_gains from protea.core.transitions_export._queries import WindowQueries, queries_of from protea.core.transitions_export._queries import write_table as write_queries from protea.core.transitions_export._residue_refs import refs_of @@ -91,7 +93,7 @@ #: the refusal can say "declared but not written yet" instead of "unknown", #: which are different answers for whoever reads the error. IMPLEMENTED: frozenset[str] = frozenset( - {"window_terms_out", "queries", "terms", "term_ancestors", "residue_refs"} + {"window_terms_out", "queries", "terms", "term_ancestors", "residue_refs", "gains"} ) #: Tables that depend on NO window, so a run asking only for these needs no @@ -104,7 +106,13 @@ #: Tables that need the axis read. ``window_terms_out`` is keyed by a window #: and still needs it, which is why this is not the complement of #: :data:`AXIS_ONLY`. -_NEEDS_AXIS: frozenset[str] = frozenset({"window_terms_out", "terms", "term_ancestors"}) +_NEEDS_AXIS: frozenset[str] = frozenset({"window_terms_out", "terms", "term_ancestors", "gains"}) + +#: Tables whose population IS the query population, so asking for either makes +#: it worth computing once. ``gains`` does not write ``queries`` and still +#: needs its population: a gain for a protein ``queries`` excludes would be a +#: gain for a query the loader does not know. +_NEEDS_QUERIES: frozenset[str] = frozenset({"queries", "gains"}) class ExportTransitionsDatasetPayload(ProteaPayload, frozen=True): @@ -140,14 +148,26 @@ class ExportTransitionsDatasetPayload(ProteaPayload, frozen=True): @field_validator("windows") @classmethod def _named_and_distinct(cls, value: list[str]) -> list[str]: - """Refuse repeats. Whether it may be EMPTY depends on the tables.""" + """Refuse repeats, and SORT, which is load-bearing and was not here. + + ``gains`` builds its rows per window as the pass walks them and never + sorts 9,6 million rows afterwards, so its bytes were a function of the + order the CALLER listed the windows in: the same 65 windows in a + different payload order gave a different sha. My own docstring for that + writer claimed "the caller walks the windows sorted" while the pass + iterated the payload as given. + + Sorting here makes the canonical order a property of the RUN rather + than of whoever wrote the list, which is what the sha has to be about. + It changes nothing semantically: the field declares a SET of windows. + """ repetidas = sorted({w for w in value if value.count(w) > 1}) if repetidas: raise ValueError( f"windows repeats {repetidas}. A window exported twice would write " "its terms twice and the row count would stop being a count" ) - return value + return sorted(value) @model_validator(mode="after") def _scope_and_vectors_match_the_tables(self) -> ExportTransitionsDatasetPayload: @@ -296,6 +316,11 @@ class _Computed(NamedTuple): terms: list[WindowTerms] queries: list[WindowQueries] + gains: list[WindowGains] + #: The six parallel lists ``gains`` accumulates into. Built once and + #: appended per window, because a tuple per row over 9,6 million rows is + #: the one avoidable cost in that table. + gain_columns: dict[str, list] def _emit_terms(r: WindowTerms, eset: Any, emit: EmitFn) -> WindowTerms: @@ -343,6 +368,26 @@ def _emit_queries(q: WindowQueries, facts: WindowFacts, emit: EmitFn) -> WindowQ return q +def _emit_gains(g: WindowGains, emit: EmitFn) -> WindowGains: + emit( + "transitions.gains", + f"{g.window[:8]}: {g.rows} gains over {g.distinct_terms} terms, {g.closure} closure", + { + "window": g.window, + "gains": g.rows, + # Per stratum, so a bundle whose strata shifted is visible from the + # event instead of from a diff of two parquets. + **{f"gains_{k}": v for k, v in g.by_stratum.items()}, + "distinct_terms": g.distinct_terms, + # Which graph the gain is expressed in. The same window under two + # closures is two different truths. + "closure": g.closure, + }, + "info", + ) + return g + + def _refuse_disagreeing_windows( hecho: _Computed, p: ExportTransitionsDatasetPayload, emit: EmitFn ) -> None: @@ -376,6 +421,50 @@ def _refuse_disagreeing_windows( ) +def _write_all( + store: Any, + p: ExportTransitionsDatasetPayload, + tmp: str, + lote: _Bundle, + vector_uri: str | None, +) -> dict[str, dict[str, Any]]: + """Write every requested table into ``tmp`` and publish each one. + + One function per table would read better and would also mean six more + signatures taking the same five things. This stays one place where the set + of tables a run writes can be read off in order. + """ + hecho, axis = lote.hecho, lote.axis + tablas: dict[str, dict[str, Any]] = {} + if "window_terms_out" in p.tables and axis is not None: + tablas["window_terms_out"] = _write_one( + store, p, tmp, "window_terms_out", lambda path: write_table(hecho.terms, path) + ) + if "queries" in p.tables: + tablas["queries"] = _write_one( + store, p, tmp, "queries", lambda path: write_queries(hecho.queries, path) + ) + if "gains" in p.tables: + tablas["gains"] = _write_one( + store, p, tmp, "gains", lambda path: write_gains(hecho.gain_columns, path) + ) + if "terms" in p.tables and axis is not None and vector_uri is not None: + sha = p.term_vectors_sha256 or "" + tablas["terms"] = _write_one( + store, p, tmp, "terms", lambda path: write_terms(axis, vector_uri, sha, path) + ) + if "term_ancestors" in p.tables and axis is not None: + tablas["term_ancestors"] = _write_one( + store, p, tmp, "term_ancestors", lambda path: write_ancestors(axis, path) + ) + if "residue_refs" in p.tables and lote.refs is not None: + refs = lote.refs + tablas["residue_refs"] = _write_one( + store, p, tmp, "residue_refs", lambda path: write_refs(refs, path) + ) + return tablas + + class ExportTransitionsDatasetOperation(Operation): name = "export_transitions_dataset" description = ( @@ -408,7 +497,6 @@ def execute( }, "info", ) - vivos = set(axis.aspect) if axis is not None else set() refs = ( refs_of(session, uuid.UUID(p.codes_config_id)) if "residue_refs" in p.tables and p.codes_config_id @@ -429,7 +517,7 @@ def execute( }, "info", ) - hecho = self._per_window(session, p, vivos, running, emit) + hecho = self._per_window(session, p, axis, running, emit) _refuse_disagreeing_windows(hecho, p, emit) return self._publish(p, _Bundle(hecho=hecho, axis=axis, refs=refs), running, payload, emit) @@ -437,12 +525,12 @@ def _per_window( self, session: Session, p: ExportTransitionsDatasetPayload, - axis: set[str], + axis: Axis | None, running: str, emit: EmitFn, ) -> _Computed: """Check and compute every window before anything is written.""" - hecho = _Computed(terms=[], queries=[]) + hecho = _Computed(terms=[], queries=[], gains=[], gain_columns=new_columns()) for raw in p.windows: eset = _window_or_refuse(session, raw) _refuse_wrong_pivot(eset, p.cut, f"window {raw}") @@ -452,11 +540,17 @@ def _per_window( # slightly different reads. facts = facts_of(session, eset) refuse_after_the_cut(facts, p.cut_version) - if "window_terms_out" in p.tables: - hecho.terms.append(_emit_terms(terms_out(session, facts, axis), eset, emit)) - if "queries" in p.tables: - hecho.queries.append( - _emit_queries(queries_of(session, facts, p.categories), facts, emit) + if "window_terms_out" in p.tables and axis is not None: + hecho.terms.append( + _emit_terms(terms_out(session, facts, set(axis.aspect)), eset, emit) + ) + q = queries_of(session, facts, p.categories) if _NEEDS_QUERIES & set(p.tables) else None + if q is not None and "queries" in p.tables: + hecho.queries.append(_emit_queries(q, facts, emit)) + if q is not None and "gains" in p.tables and axis is not None: + poblacion = {r.query for r in q.rows} + hecho.gains.append( + _emit_gains(gains_of(facts, axis.aspect, poblacion, hecho.gain_columns), emit) ) return hecho @@ -470,8 +564,6 @@ def _publish( ) -> OperationResult: """Write every requested table, publish it, and report its digest.""" store = get_artifact_store(load_settings(Path(__file__).resolve().parents[3])) - hecho, axis = lote.hecho, lote.axis - tablas: dict[str, dict[str, Any]] = {} # Verified BEFORE anything is written: those two columns are the reason # the vectors are carried by reference at all, and a sha nobody checked # is worse in a table than absent from one. @@ -481,28 +573,7 @@ def _publish( else None ) with tempfile.TemporaryDirectory(prefix="transitions-") as tmp: - if "window_terms_out" in p.tables: - tablas["window_terms_out"] = _write_one( - store, p, tmp, "window_terms_out", lambda path: write_table(hecho.terms, path) - ) - if "queries" in p.tables: - tablas["queries"] = _write_one( - store, p, tmp, "queries", lambda path: write_queries(hecho.queries, path) - ) - if "terms" in p.tables and axis is not None and vector_uri is not None: - sha = p.term_vectors_sha256 or "" - tablas["terms"] = _write_one( - store, p, tmp, "terms", lambda path: write_terms(axis, vector_uri, sha, path) - ) - if "term_ancestors" in p.tables and axis is not None: - tablas["term_ancestors"] = _write_one( - store, p, tmp, "term_ancestors", lambda path: write_ancestors(axis, path) - ) - if "residue_refs" in p.tables and lote.refs is not None: - refs = lote.refs - tablas["residue_refs"] = _write_one( - store, p, tmp, "residue_refs", lambda path: write_refs(refs, path) - ) + tablas = _write_all(store, p, tmp, lote, vector_uri) result = { "tables": tablas, "windows": len(p.windows), diff --git a/protea/core/transitions_export/_gains.py b/protea/core/transitions_export/_gains.py new file mode 100644 index 00000000..aece6541 --- /dev/null +++ b/protea/core/transitions_export/_gains.py @@ -0,0 +1,191 @@ +"""The truth of a window: which terms each query gained, with its stratum. + +THIS IS THE TABLE THAT CARRIES THE TRUTH, so the two rules that decide what a +gain IS are read from where the evaluation already applies them rather than +restated: the three scored maps come out of the window's own verified parquet, +already trimmed by the terms of interest, and the aspect comes from the same +axis the evaluation classified under. + +THE POPULATION IS THE ONE ``queries`` HAS, and that is a guard rather than a +filter. ``queries`` excludes a protein with no stored sequence, because the +exported FASTA excludes it and the training rows have to be about the scored +population. A gain for a protein outside that set would be a gain for a query +the loader does not know, and the two tables would disagree about who the +queries are without either being wrong on its own. Measured 2026-11-08 over the +65 admissible windows: **zero** delta proteins lack a sequence, so this guard +changes nothing today and says so when that stops being true. + +THE SHAPE, measured before the code was written. Over those 65 windows +``gains`` is **9.596.563 rows**, the largest window 691.553 and the mean +147.639. Six short columns at that count is the one place where building a +tuple per row would hurt: 9,6 million objects holding 58 million pointers. So +the columns are built as six parallel lists directly, which is why this module +looks like it does. +""" + +from __future__ import annotations + +from typing import NamedTuple + +import pyarrow as pa +import pyarrow.parquet as pq + +from protea.core.transitions_export._window_facts import WindowFacts + +#: The six columns the laboratory's loader reads. +SCHEMA = pa.schema( + [ + ("window", pa.string()), + ("query", pa.string()), + ("aspect", pa.string()), + ("go_id", pa.string()), + ("closure", pa.string()), + ("stratum", pa.string()), + ] +) + +#: The three scored strata, in the order the evaluation names them. ``removed`` +#: is not here: those proteins only LOST annotation, which the evaluation +#: reports and never scores. +STRATA: tuple[str, ...] = ("nk", "lk", "pk") + + +class WindowGains(NamedTuple): + """One window's gains, with what the guards found.""" + + window: str + closure: str + rows: int + #: Per stratum, so a bundle whose strata shifted is visible from the event + #: rather than from a diff of two parquets. + by_stratum: dict[str, int] + distinct_terms: int + + +def closure_of(facts: WindowFacts) -> str: + """Which graphs this window's two ends close in, DERIVED, or a refusal. + + IT IS NOT ``truth_profile``, AND THE FIRST VERSION OF THIS FUNCTION MADE + EXACTLY THAT MISTAKE. It mapped ``projected`` to ``pivot`` and + ``as_closed`` to ``native``, which ADR-D56 says in as many words is wrong: + ``truth_profile`` says HOW the truth is derived and ``closure`` says WHICH + GRAPHS the two ends close in, and they vary independently. + + The proof is in the database. The (A) and (B) twins of ``160 -> 177`` are + BOTH ``projected`` and differ only in their natives, so under the mapping + the (B) one, ``e84dfec8``, would have been written as ``pivot``. The + desktop session caught it by reading the ADR that this project's own + earlier PR had written, and it had made the same mistake a day before and + withdrawn it. + + So the value comes from comparing the window's two natives against its + pivot and against the ontology each end's corpus was loaded under, the way + the window inventory already derives it, and a combination that is neither + is refused rather than guessed. Measured over the 131 evaluation sets: + 122 are ``native`` and 9 are ``pivot``. + """ + eset = facts.eset + pivote = eset.pivot_snapshot_id + viejo, nuevo = eset.old_native_snapshot_id, eset.new_native_snapshot_id + if viejo == pivote and nuevo == pivote: + return "pivot" + if viejo == facts.old_end_snapshot and nuevo == facts.new_end_snapshot: + return "native" + raise ValueError( + f"window {facts.window} has natives {str(viejo)[:8]} and {str(nuevo)[:8]} " + f"against pivot {str(pivote)[:8]} and corpus ontologies " + f"{str(facts.old_end_snapshot)[:8]} and {str(facts.new_end_snapshot)[:8]}: " + "neither both-on-the-pivot nor each-on-its-own-era. The closure column says " + "which graphs the two ends close in, and a third combination has no name yet, " + "so it is refused rather than written as one of the two" + ) + + +def gains_of( + facts: WindowFacts, aspect: dict[str, str], population: set[str], columns: dict[str, list] +) -> WindowGains: + """Append one window's gains to the shared columns, and say what it added. + + It appends rather than returning rows: at 9,6 million rows over 65 windows, + a list of tuples per window and a concatenation afterwards is the one + avoidable cost in this table. + """ + cierre = closure_of(facts) + por_estrato: dict[str, int] = {} + terminos: set[str] = set() + fuera: set[str] = set() + sin_aspecto: set[str] = set() + for estrato in STRATA: + n = 0 + for query, gos in sorted((getattr(facts.data, estrato, {}) or {}).items()): + if query not in population: + fuera.add(query) + continue + for go in sorted(gos): + if go not in aspect: + sin_aspecto.add(go) + continue + columns["window"].append(facts.window) + columns["query"].append(query) + columns["aspect"].append(aspect[go]) + columns["go_id"].append(go) + columns["closure"].append(cierre) + columns["stratum"].append(estrato) + terminos.add(go) + n += 1 + por_estrato[estrato] = n + _refuse_gaps(facts, fuera, sin_aspecto) + return WindowGains( + window=facts.window, + closure=cierre, + rows=sum(por_estrato.values()), + by_stratum=por_estrato, + distinct_terms=len(terminos), + ) + + +def _refuse_gaps(facts: WindowFacts, fuera: set[str], sin_aspecto: set[str]) -> None: + """A gain outside the query population, or on a term the axis cannot name. + + BOTH ARE GUARDS AND NEITHER FILTERS ANYTHING TODAY, measured. They refuse + rather than drop because dropping would make ``gains`` quietly smaller than + the truth it is a copy of, and a row count is the only thing a reader could + compare it against. + """ + if fuera: + raise ValueError( + f"window {facts.window}: {len(fuera)} proteins have gains and are not in " + f"the query population: {sorted(fuera)[:5]}. queries excludes a protein " + "with no stored sequence because the scored FASTA does, so a gain for one " + "would be a gain for a query the loader does not know" + ) + if sin_aspecto: + raise ValueError( + f"window {facts.window}: {len(sin_aspecto)} gained terms are not on the " + f"declared axis: {sorted(sin_aspecto)[:5]}. The loader groups by aspect, " + "so a term the axis cannot name is a row it cannot place" + ) + + +def new_columns() -> dict[str, list]: + """The six parallel lists, empty, in the schema's order.""" + return {name: [] for name in SCHEMA.names} + + +def write_table(columns: dict[str, list], path: str) -> int: + """Write the accumulated columns, already in a stable order. + + The order is stable by construction and not by a final sort: the payload's + windows are SORTED BY ITS OWN VALIDATOR, and within a window the strata go + in their declared order with each query and each term sorted. Sorting 9,6 + million rows again here would cost more than the property is worth when the + property is already there. + + THE SORT IN THE VALIDATOR IS WHAT MAKES THAT TRUE, and it was not there + when this docstring first claimed it: the pass walked ``p.windows`` as + given, so the same 65 windows listed in a different order produced a + different sha. The property a digest has to have is about the run, not + about whoever wrote the list. + """ + pq.write_table(pa.table(columns, schema=SCHEMA), path) + return len(columns["window"]) diff --git a/protea/core/transitions_export/_window_facts.py b/protea/core/transitions_export/_window_facts.py index da23f944..31675cf3 100644 --- a/protea/core/transitions_export/_window_facts.py +++ b/protea/core/transitions_export/_window_facts.py @@ -43,10 +43,16 @@ class WindowFacts(NamedTuple): version_t: str version_t_prime: str horizon_days: int + #: The ontology each end's CORPUS was loaded under, which is what the + #: closure is derived by comparing the window's two natives against. Read + #: from the same annotation_set rows the versions come from, so it costs + #: no extra query. + old_end_snapshot: Any + new_end_snapshot: Any -def _end(session: Session, annotation_set_id: Any, lado: str, window: Any) -> tuple[str, date]: - """One end's release and publication date, both validated. +def _end(session: Session, annotation_set_id: Any, lado: str, window: Any) -> tuple[str, date, Any]: + """One end's release, publication date and corpus ontology, all validated. It returns the VALUES and not the row so the caller's types narrow: a check inside a different function does not narrow ``str | None`` at the call @@ -68,7 +74,7 @@ def _end(session: Session, annotation_set_id: Any, lado: str, window: Any) -> tu "publication date, so the horizon cannot be computed. Run " "refresh_goa_release_dates first" ) - return aset.source_version, aset.source_published_at.date() + return aset.source_version, aset.source_published_at.date(), aset.ontology_snapshot_id def refuse_after_the_cut(facts: WindowFacts, cut_version: str) -> None: @@ -109,8 +115,8 @@ def facts_of(session: Session, eset: EvaluationSet) -> WindowFacts: """ from protea.core.evaluation import load_evaluation_data_for_set - version_t, fecha_t = _end(session, eset.old_annotation_set_id, "old", eset.id) - version_t1, fecha_t1 = _end(session, eset.new_annotation_set_id, "new", eset.id) + version_t, fecha_t, snap_t = _end(session, eset.old_annotation_set_id, "old", eset.id) + version_t1, fecha_t1, snap_t1 = _end(session, eset.new_annotation_set_id, "new", eset.id) horizonte = (fecha_t1 - fecha_t).days if horizonte <= 0: raise ValueError( @@ -127,4 +133,6 @@ def facts_of(session: Session, eset: EvaluationSet) -> WindowFacts: version_t=version_t, version_t_prime=version_t1, horizon_days=horizonte, + old_end_snapshot=snap_t, + new_end_snapshot=snap_t1, ) diff --git a/tests/test_payload_validation.py b/tests/test_payload_validation.py index a9fcdce2..788cefc0 100644 --- a/tests/test_payload_validation.py +++ b/tests/test_payload_validation.py @@ -629,7 +629,7 @@ # que se pidio no se distingue de uno completo. "export_transitions_dataset", ExportTransitionsDatasetPayload, - {"windows": ["w"], "cut": "x", "cut_version": "211", "tables": ["gains"]}, + {"windows": ["w"], "cut": "x", "cut_version": "211", "tables": ["query_state"]}, ("tables",), ), ( diff --git a/tests/test_the_axis_terms_a_window_leaves_out.py b/tests/test_the_axis_terms_a_window_leaves_out.py index 38c1a3b7..a44cd852 100644 --- a/tests/test_the_axis_terms_a_window_leaves_out.py +++ b/tests/test_the_axis_terms_a_window_leaves_out.py @@ -80,6 +80,8 @@ def _facts(eset=None, data=None, version_t="200", version_t_prime="201", dias=30 version_t=version_t, version_t_prime=version_t_prime, horizon_days=dias, + old_end_snapshot=uuid.uuid4(), + new_end_snapshot=uuid.uuid4(), ) @@ -178,8 +180,11 @@ class TestWhatTheOperationRefuses: def test_a_table_declared_by_the_adr_but_not_written_yet(self): """Refused and not skipped, and the message says which of the two it is.""" with pytest.raises(ValueError, match="does not write them yet"): + # ``gains`` landed, so the example moved to one of the four that + # have not: a test whose subject gets implemented has to move with + # it or it stops testing the refusal. ExportTransitionsDatasetPayload( - windows=["w"], cut="c", cut_version="211", tables=["gains"] + windows=["w"], cut="c", cut_version="211", tables=["query_state"] ) def test_a_table_no_adr_declares(self): @@ -221,21 +226,30 @@ def test_naming_the_revision_admits_it_and_a_boolean_would_not(self): assert not hasattr(ExportTransitionsDatasetPayload, "accept_any_revision") -def test_the_ten_tables_are_declared_and_five_are_written(): +def test_the_ten_tables_are_declared_and_six_are_written(): """The gap between the two lists is the honest state of this operation. Hard-coded on purpose: a table landing has to come with a line here, so - the list cannot drift into claiming more than is written. + the list cannot drift into claiming more than is written. The four left + are ``query_state``, ``neighbourhood``, ``neighbour_state`` and + ``feature_tokens``. """ assert len(TABLES) == 10 assert IMPLEMENTED == { "window_terms_out", "queries", + "gains", "terms", "term_ancestors", "residue_refs", } assert IMPLEMENTED <= set(TABLES) + assert set(TABLES) - IMPLEMENTED == { + "query_state", + "neighbourhood", + "neighbour_state", + "feature_tokens", + } def test_the_operation_says_it_writes_no_dataset_row(): diff --git a/tests/test_the_queries_a_window_asks_about.py b/tests/test_the_queries_a_window_asks_about.py index 85d5a797..09c7e4f3 100644 --- a/tests/test_the_queries_a_window_asks_about.py +++ b/tests/test_the_queries_a_window_asks_about.py @@ -31,10 +31,13 @@ class _Aset: - def __init__(self, version="200", fecha=datetime(2022, 1, 1)) -> None: + def __init__(self, version="200", fecha=datetime(2022, 1, 1), snapshot=None) -> None: self.id = uuid.uuid4() self.source_version = version self.source_published_at = fecha + #: The ontology this corpus was loaded under. The closure is derived + #: by comparing the window's natives against it, so the facts carry it. + self.ontology_snapshot_id = snapshot if snapshot is not None else uuid.uuid4() class _Eset: @@ -92,6 +95,8 @@ def _facts(version_t="200", version_t_prime="201", dias=31, data=None): version_t=version_t, version_t_prime=version_t_prime, horizon_days=dias, + old_end_snapshot=uuid.uuid4(), + new_end_snapshot=uuid.uuid4(), ) @@ -271,9 +276,12 @@ class TestTheTwoTablesAgreeAboutTheirWindows: def _computed(self, ventanas_terminos, ventanas_consultas): from protea.core.operations.export_transitions_dataset import _Computed + from protea.core.transitions_export._gains import new_columns from protea.core.transitions_export._window_terms_out import WindowTerms return _Computed( + gains=[], + gain_columns=new_columns(), terms=[ WindowTerms( window=w, diff --git a/tests/test_the_truth_a_window_carries.py b/tests/test_the_truth_a_window_carries.py new file mode 100644 index 00000000..c93a3c87 --- /dev/null +++ b/tests/test_the_truth_a_window_carries.py @@ -0,0 +1,281 @@ +"""``gains``: which terms each query gained, with its stratum and its closure. + +THIS IS THE TABLE THAT CARRIES THE TRUTH, so the subject of this file is the +two ways it could quietly stop being a copy of it. + +ONE: its population. ``queries`` excludes a protein with no stored sequence, +because the exported FASTA excludes it and the training rows have to be about +the scored population. A gain for a protein outside that set would be a gain +for a query the loader does not know, and the two tables would disagree about +who the queries are without either being wrong on its own. Measured 2026-11-08 +over the 65 admissible windows: **zero** delta proteins lack a sequence, so the +guard changes nothing today and says so when that stops being true. + +TWO: its closure. ``projected`` means the gain is expressed on the pivot and +``as_closed`` means on each end's own graph, and the same window under two +closures is two different truths. The mapping lives in one place and a profile +with no entry is refused rather than guessed. + +THE SHAPE, measured before the code: 9.596.563 rows over those 65 windows, the +largest window 691.553. That is the one place where a tuple per row would hurt, +so the columns are six parallel lists. +""" + +from __future__ import annotations + +import uuid +from pathlib import Path + +import pyarrow.parquet as pq +import pytest + +from protea.core.transitions_export import _gains as G +from protea.core.transitions_export._window_facts import WindowFacts + +ASPECTO = {"GO:1": "BPO", "GO:2": "MFO", "GO:3": "CCO"} + + +PIVOTE = uuid.uuid4() +CORPUS_T = uuid.uuid4() +CORPUS_T1 = uuid.uuid4() + + +class _Eset: + def __init__(self, perfil="projected", viejo=None, nuevo=None) -> None: + self.id = uuid.uuid4() + self.truth_profile = perfil + self.pivot_snapshot_id = PIVOTE + # Por defecto (A): los dos nativos en el pivote. + self.old_native_snapshot_id = PIVOTE if viejo is None else viejo + self.new_native_snapshot_id = PIVOTE if nuevo is None else nuevo + + +class _Data: + def __init__(self, nk=None, lk=None, pk=None, removed=None) -> None: + self.nk = nk or {} + self.lk = lk or {} + self.pk = pk or {} + self.removed = removed or {} + + +def _facts(data, perfil="projected", viejo=None, nuevo=None): + e = _Eset(perfil, viejo, nuevo) + return WindowFacts( + eset=e, + window=str(e.id), + data=data, + version_t="200", + version_t_prime="201", + horizon_days=31, + old_end_snapshot=CORPUS_T, + new_end_snapshot=CORPUS_T1, + ) + + +# ------------------------------------------------------------------ el cierre + + +class TestTheClosure: + """IT IS DERIVED FROM THE NATIVES, AND THE FIRST VERSION GOT IT WRONG. + + It mapped ``projected`` to ``pivot`` and ``as_closed`` to ``native``, which + ADR-D56 says in as many words is wrong: ``truth_profile`` says HOW the + truth is derived and ``closure`` says WHICH GRAPHS the two ends close in, + and they vary independently. + + So these tests walk combinations of NATIVES and not values of + ``TRUTH_PROFILES``, and the one that matters is + ``test_two_projected_windows_can_have_different_closures``: the (A) and (B) + twins of ``160 -> 177`` in the database are both ``projected`` and differ + only there, so under the mapping the (B) one would have been written as + ``pivot``. The desktop session caught it by reading the ADR this project's + own earlier PR had written. + """ + + def test_both_natives_on_the_pivot_is_pivot(self): + assert G.closure_of(_facts(_Data())) == "pivot" + + def test_each_native_on_its_own_era_is_native(self): + assert G.closure_of(_facts(_Data(), viejo=CORPUS_T, nuevo=CORPUS_T1)) == "native" + + def test_two_projected_windows_can_have_DIFFERENT_closures(self): + """THE TEST THIS CLASS EXISTS FOR, and the ADR's own proof.""" + a = _facts(_Data(), perfil="projected") + b = _facts(_Data(), perfil="projected", viejo=CORPUS_T, nuevo=CORPUS_T1) + + assert a.eset.truth_profile == b.eset.truth_profile == "projected" + assert G.closure_of(a) == "pivot" + assert G.closure_of(b) == "native" + + def test_as_closed_on_the_pivot_is_still_pivot(self): + """The other direction of the same independence.""" + assert G.closure_of(_facts(_Data(), perfil="as_closed")) == "pivot" + + def test_a_third_combination_is_refused(self): + """Neither both-on-the-pivot nor each-on-its-own-era has a name yet.""" + with pytest.raises(ValueError, match="neither both-on-the-pivot"): + G.closure_of(_facts(_Data(), viejo=PIVOTE, nuevo=uuid.uuid4())) + + def test_the_profile_is_not_read_at_all(self): + import inspect + + fuente = inspect.getsource(G.closure_of) + + assert "truth_profile" not in fuente.split('"""')[2] + + +# ------------------------------------------------------------- lo que escribe + + +def test_it_writes_one_row_per_query_and_term_with_its_stratum(): + cols = G.new_columns() + data = _Data(nk={"P1": {"GO:2", "GO:1"}}, pk={"P2": {"GO:3"}}) + + g = G.gains_of(_facts(data), ASPECTO, {"P1", "P2"}, cols) + + assert g.rows == 3 + assert g.by_stratum == {"nk": 2, "lk": 0, "pk": 1} + assert g.distinct_terms == 3 + assert g.closure == "pivot" + # Orden estable por construccion: estratos declarados, consulta ordenada, + # termino ordenado. Sin un sort final de 9,6 millones de filas. + assert cols["go_id"] == ["GO:1", "GO:2", "GO:3"] + assert cols["stratum"] == ["nk", "nk", "pk"] + assert cols["aspect"] == ["BPO", "MFO", "CCO"] + assert cols["closure"] == ["pivot", "pivot", "pivot"] + + +def test_removed_is_not_a_stratum(): + """Those proteins only LOST annotation: reported by the evaluation, never scored.""" + cols = G.new_columns() + data = _Data(nk={"P1": {"GO:1"}}, removed={"P9": {"GO:2"}}) + + g = G.gains_of(_facts(data), ASPECTO, {"P1", "P9"}, cols) + + assert g.rows == 1 + assert "removed" not in G.STRATA + assert "P9" not in cols["query"] + + +def test_several_windows_accumulate_into_the_same_columns(): + """Appending rather than returning rows is the measured decision.""" + cols = G.new_columns() + a = G.gains_of(_facts(_Data(nk={"P1": {"GO:1"}})), ASPECTO, {"P1"}, cols) + b = G.gains_of(_facts(_Data(nk={"P2": {"GO:2"}})), ASPECTO, {"P2"}, cols) + + assert len(cols["window"]) == 2 + assert {a.window, b.window} == set(cols["window"]) + + +# ------------------------------------------------------------------ los guardas + + +class TestWhatItRefusesRatherThanDrops: + """Both guards refuse rather than drop, and neither filters anything today. + + Dropping would make ``gains`` quietly smaller than the truth it copies, and + a row count is the only thing a reader could compare it against. + """ + + def test_a_gain_for_a_protein_queries_excludes(self): + """It would be a gain for a query the loader does not know.""" + cols = G.new_columns() + data = _Data(nk={"P1": {"GO:1"}}, pk={"P_sin_secuencia": {"GO:2"}}) + + with pytest.raises(ValueError, match="not in the query population"): + G.gains_of(_facts(data), ASPECTO, {"P1"}, cols) + + def test_a_gain_on_a_term_the_axis_cannot_name(self): + """The loader groups by aspect, so it is a row it cannot place.""" + cols = G.new_columns() + data = _Data(nk={"P1": {"GO:1", "GO:9999"}}) + + with pytest.raises(ValueError, match="not on the declared axis"): + G.gains_of(_facts(data), ASPECTO, {"P1"}, cols) + + def test_the_message_names_the_window_and_a_sample(self): + cols = G.new_columns() + facts = _facts(_Data(nk={"P_fuera": {"GO:1"}})) + + with pytest.raises(ValueError) as exc: + G.gains_of(facts, ASPECTO, set(), cols) + + assert facts.window in str(exc.value) + assert "P_fuera" in str(exc.value) + + +# ------------------------------------------------------------------- el parquet + + +def test_the_six_columns_are_the_loader_s(): + assert G.SCHEMA.names == ["window", "query", "aspect", "go_id", "closure", "stratum"] + + +def test_the_table_is_written_from_the_accumulated_columns(tmp_path: Path): + cols = G.new_columns() + G.gains_of(_facts(_Data(nk={"P1": {"GO:1"}}, lk={"P2": {"GO:2"}})), ASPECTO, {"P1", "P2"}, cols) + destino = tmp_path / "gains.parquet" + + assert G.write_table(cols, str(destino)) == 2 + + tabla = pq.read_table(destino).to_pydict() + assert tabla["query"] == ["P1", "P2"] + assert tabla["stratum"] == ["nk", "lk"] + + +def test_an_empty_run_writes_an_empty_table_rather_than_failing(tmp_path: Path): + """A window set with no gains is a real answer, and the sha describes it.""" + destino = tmp_path / "vacio.parquet" + + assert G.write_table(G.new_columns(), str(destino)) == 0 + assert pq.read_table(destino).num_rows == 0 + + +def test_the_same_windows_in_any_payload_order_give_the_same_sha(tmp_path: Path): + """THE PROPERTY THE SORT IN THE VALIDATOR EXISTS FOR. + + ``gains`` never sorts its rows at the end, so without a canonical window + order its bytes were a function of how the caller listed them. My own + docstring claimed "the caller walks the windows sorted" while the pass + iterated the payload as given, which is the same shape of mistake as a test + written from the implementation: the sentence was about what should happen. + """ + import hashlib + + from protea.core.operations.export_transitions_dataset import ( + ExportTransitionsDatasetPayload, + ) + + a, b, c = (str(uuid.uuid4()) for _ in range(3)) + uno = ExportTransitionsDatasetPayload( + windows=[c, a, b], cut="x", cut_version="211", tables=["gains"] + ) + otro = ExportTransitionsDatasetPayload( + windows=[b, c, a], cut="x", cut_version="211", tables=["gains"] + ) + + assert uno.windows == otro.windows == sorted([a, b, c]) + + # Y los bytes, con la misma poblacion escrita en los dos ordenes. + def escribe(orden, destino): + cols = G.new_columns() + for nombre in orden: + e = _Eset() + e.id = nombre + facts = WindowFacts( + eset=e, + window=str(nombre), + data=_Data(nk={"P1": {"GO:1"}}), + version_t="200", + version_t_prime="201", + horizon_days=31, + old_end_snapshot=CORPUS_T, + new_end_snapshot=CORPUS_T1, + ) + G.gains_of(facts, ASPECTO, {"P1"}, cols) + G.write_table(cols, str(destino)) + return hashlib.sha256(destino.read_bytes()).hexdigest() + + assert escribe(uno.windows, tmp_path / "a.parquet") == ( + escribe(otro.windows, tmp_path / "b.parquet") + )