From e28e8e5b1308b742ef7b59cfaf291fc521dbaa55 Mon Sep 17 00:00:00 2001 From: Jorge Rivera Date: Sun, 25 Oct 2026 00:29:41 +0200 Subject: [PATCH] fix(dataset-store): rewrite stored warnings that name refs into counted messages Migration 0039 rewrites each warning in `datasets.version_record.warnings` that #129's writer no longer produces: a `coverage_below_scope` sentence listing unserved entities or group members, and a `currency_unmapped_omitted` or `deflator_missing_omitted` list of entities. Each becomes the counted message and `{count, refs}` field the current writer stores. A range-window member sentence counted off-cadence and out-of-window members together, and the row stores nothing that splits them, so it becomes a counted sentence true of both with no field. Only the warnings column changes, and a rerun touches zero rows. --- .../migrations/0039_warning_counted_refs.sql | 253 ++++++++++++++++++ tests/dataset_store/_warning_refs.py | 17 +- tests/dataset_store/test_live.py | 112 +++++++- tests/dataset_store/test_migrate.py | 1 + tests/platform/test_migrations_immutable.py | 3 + 5 files changed, 384 insertions(+), 2 deletions(-) create mode 100644 src/dataset_store/migrations/0039_warning_counted_refs.sql diff --git a/src/dataset_store/migrations/0039_warning_counted_refs.sql b/src/dataset_store/migrations/0039_warning_counted_refs.sql new file mode 100644 index 00000000..75721c4e --- /dev/null +++ b/src/dataset_store/migrations/0039_warning_counted_refs.sql @@ -0,0 +1,253 @@ +-- Rewrites the stored warnings that name refs in their message into the shape +-- `materialise.finish` writes: a message that counts, and a structured field +-- that carries the refs for a reader to label (`materialise.codes.CountedRefs`, +-- `materialise.codes.CountedEntityYears`). +-- +-- `datasets.version_record.warnings` is the column every read assembles a +-- manifest's warnings from (`datasets_api.manifest.stored_manifest`), and +-- `record_version` inserts with ON CONFLICT DO NOTHING, so a later mint of the +-- same content version never replaces it. `immutable_body` keeps its copy as +-- stored: no read decodes its warnings once the facts are filled. The trigger +-- `version_record_fill_facts` fires on INSERT only, so the UPDATE fires +-- nothing. Description records copy no text from these codes. Content +-- versions do not move: warnings are outside the version preimage. +-- +-- Each old sentence follows the catalogue detail, which is kept as stored: +-- +-- coverage_below_scope, unserved entities: +-- ` No observations were served for KEN, UGA and 3 more.` +-- The first five sorted bare entity dcids, with a count of the rest. It +-- becomes ` No observations were served for 5 requested entities.` and +-- `unserved_entities` `{count, refs}`, each ref `dcid:` prefixed, in the +-- stored order. A row stored while member sentences already counted can +-- hold counted member sentences after it, which stay as stored. +-- coverage_below_scope, group members that served nothing, under a window +-- that is not a range: +-- ` No observations were served for dcid:A, dcid:B, which hold none at the +-- observation period the other members serve.` +-- The first five `dcid:` StatVar refs in request order, with a count of the +-- rest. Under such a window every uncounted member is off cadence +-- (`materialise.generation._plan_group_members`), so it becomes ` 2 of 7 +-- series report at a different frequency and are left out of the +-- figures.`, `off_cadence_members` `{count, refs}` and `member_count`. +-- coverage_below_scope, group members that served nothing, under a range +-- window: +-- ` No observations were served for dcid:A, dcid:B, which hold none at the +-- observation period the other members serve in the requested years.` +-- This sentence counted two kinds together: members whose facets at another +-- observation period hold the requested years, which the writer now calls +-- `off_cadence_members`, and members holding no observation in those years, +-- its `out_of_window_members`. The row records nothing that tells them +-- apart: the split was computed from each member's probe candidates and +-- never stored. Either field would claim a split the data cannot back, so +-- the sentence becomes the statement true of both, ` 2 of 7 series hold no +-- observations at the group's observation period in the requested years +-- and are left out of the figures.`, with no structured field and no +-- `member_count`. The refs stay only in `immutable_body`. +-- In both, the series total is the length of the row's +-- `trace.resolved_stat_vars`, one entry per member request +-- (`materialise.finish._finish_group`), the requests the writer counts. A +-- trace holding no such list leaves the total out: ` 2 series report ...`, +-- ` 2 series hold ...`, with no `member_count`. +-- Either can be followed by the countries-fallback sentence, kept as stored. +-- currency_unmapped_omitted: +-- ` 12 entities: A, B, ..., J (truncated)`, no closing full stop. The +-- first ten sorted bare entity dcids. It becomes ` It left out 12 +-- entities.` and `omitted_entities` `{count, refs}`. +-- deflator_missing_omitted: +-- ` 2 entities: A (1990-2001, 2004), B (2010)[ (truncated)]`. The first +-- ten entities, sorted, each with every run of its omitted years. It +-- becomes ` It left out rows of 2 entities.` and `omitted_entity_years` +-- `{count, entities}`, each entity `{ref, years}`. +-- +-- A warning in none of these shapes is returned as stored, which covers every +-- warning already in the new shape: a new unserved sentence counts "requested +-- entities" where the old one lists dcids, and the new member and conversion +-- sentences name no ref. A row is updated only when one of its warnings +-- changes, so a repeat run touches zero rows. +-- +-- This migration must run only once no serving revision writes the old +-- shapes: the image that writes the new shapes is the previous image of the +-- deploy that carries it. + +CREATE FUNCTION datasets.warning_counted_refs_0039(warning jsonb, member_count integer) +RETURNS jsonb +LANGUAGE plpgsql IMMUTABLE AS $rewrite$ +DECLARE + fallback CONSTANT text := + ' No place was named, so the answer covers the countries with data, ' + 'because the full list of places with data was too large or unavailable.'; + truncated CONSTANT text := ' (truncated)'; + message text := warning ->> 'message'; + tail text := ''; + found text[]; + named text[]; + counted integer; +BEGIN + IF warning ->> 'code' = 'coverage_below_scope' THEN + IF right(message, length(fallback)) = fallback THEN + message := left(message, -length(fallback)); + tail := fallback; + END IF; + found := regexp_match( + message, + '^(.*) No observations were served for (dcid:[^ ,]+(?:, dcid:[^ ,]+)*)' + '(?: and ([0-9]+) more)?, which hold none at the observation period the other ' + 'members serve( in the requested years)?\.$' + ); + IF found IS NOT NULL THEN + named := string_to_array(found[2], ', '); + counted := cardinality(named) + coalesce(found[3]::integer, 0); + message := found[1]; + tail := format( + ' %s%s series %s and are left out of the figures.', + counted, + CASE WHEN member_count IS NULL THEN '' ELSE format(' of %s', member_count) END, + CASE + WHEN found[4] IS NULL THEN 'report at a different frequency' + ELSE 'hold no observations at the group''s observation period in the ' + 'requested years' + END + ) || tail; + IF found[4] IS NULL THEN + warning := warning + || jsonb_build_object( + 'off_cadence_members', + jsonb_build_object('count', counted, 'refs', to_jsonb(named)) + ) + || jsonb_strip_nulls(jsonb_build_object('member_count', member_count)); + END IF; + END IF; + found := regexp_match( + message, + '^(.*) No observations were served for ([^ ,]+(?:, [^ ,]+)*)(?: and ([0-9]+) more)?' + '\.((?: [0-9]+ (?:of [0-9]+ )?series [^.]*\.)*)$' + ); + IF found IS NOT NULL THEN + named := string_to_array(found[2], ', '); + counted := cardinality(named) + coalesce(found[3]::integer, 0); + message := format( + '%s No observations were served for %s requested %s.%s', + found[1], + counted, + CASE WHEN counted = 1 THEN 'entity' ELSE 'entities' END, + found[4] + ); + warning := warning || jsonb_build_object( + 'unserved_entities', + jsonb_build_object( + 'count', + counted, + 'refs', + ( + SELECT jsonb_agg('dcid:' || entity ORDER BY position) + FROM unnest(named) WITH ORDINALITY AS listed (entity, position) + ) + ) + ); + END IF; + RETURN warning || jsonb_build_object('message', message || tail); + END IF; + + IF warning ->> 'code' NOT IN ('currency_unmapped_omitted', 'deflator_missing_omitted') THEN + RETURN warning; + END IF; + IF right(message, length(truncated)) = truncated THEN + message := left(message, -length(truncated)); + END IF; + + IF warning ->> 'code' = 'currency_unmapped_omitted' THEN + found := regexp_match(message, '^(.*) ([0-9]+) entities: ([^ ,]+(?:, [^ ,]+)*)$'); + IF found IS NULL THEN + RETURN warning; + END IF; + counted := found[2]::integer; + RETURN warning || jsonb_build_object( + 'message', + format( + '%s It left out %s %s.', + found[1], + counted, + CASE WHEN counted = 1 THEN 'entity' ELSE 'entities' END + ), + 'omitted_entities', + jsonb_build_object( + 'count', + counted, + 'refs', + ( + SELECT jsonb_agg('dcid:' || entity ORDER BY position) + FROM unnest(string_to_array(found[3], ', ')) + WITH ORDINALITY AS listed (entity, position) + ) + ) + ); + END IF; + + found := regexp_match( + message, + '^(.*) ([0-9]+) entities: ([^ ,]+ \([^)]*\)(?:, [^ ,]+ \([^)]*\))*)$' + ); + IF found IS NULL THEN + RETURN warning; + END IF; + counted := found[2]::integer; + RETURN warning || jsonb_build_object( + 'message', + format( + '%s It left out rows of %s %s.', + found[1], + counted, + CASE WHEN counted = 1 THEN 'entity' ELSE 'entities' END + ), + 'omitted_entity_years', + jsonb_build_object( + 'count', + counted, + 'entities', + ( + SELECT jsonb_agg( + jsonb_build_object( + 'ref', 'dcid:' || listed.item[1], + 'years', to_jsonb(string_to_array(listed.item[2], ', ')) + ) + ORDER BY listed.position + ) + FROM regexp_matches(found[3], '([^ ,]+) \(([^)]*)\)', 'g') + WITH ORDINALITY AS listed (item, position) + ) + ) + ); +END; +$rewrite$; + +UPDATE datasets.version_record AS v +SET warnings = rewritten.warnings +FROM ( + SELECT + r.subject_id, + r.content_version, + ( + SELECT jsonb_agg( + datasets.warning_counted_refs_0039( + warning.value, + CASE + WHEN jsonb_typeof(r.trace -> 'resolved_stat_vars') = 'array' + THEN jsonb_array_length(r.trace -> 'resolved_stat_vars') + END + ) + ORDER BY warning.position + ) + FROM jsonb_array_elements(r.warnings) WITH ORDINALITY AS warning (value, position) + ) AS warnings + FROM datasets.version_record AS r + WHERE jsonb_typeof(r.warnings) = 'array' + AND r.warnings @? '$[*] ? (@.code == "coverage_below_scope" + || @.code == "currency_unmapped_omitted" + || @.code == "deflator_missing_omitted")' +) AS rewritten +WHERE v.subject_id = rewritten.subject_id + AND v.content_version = rewritten.content_version + AND v.warnings IS DISTINCT FROM rewritten.warnings; + +DROP FUNCTION datasets.warning_counted_refs_0039(jsonb, integer); diff --git a/tests/dataset_store/_warning_refs.py b/tests/dataset_store/_warning_refs.py index e77da9dc..8dfc581e 100644 --- a/tests/dataset_store/_warning_refs.py +++ b/tests/dataset_store/_warning_refs.py @@ -25,6 +25,13 @@ ) ENTITIES = ["country/BDI", "country/KEN", "country/RWA", "country/SSD"] UNSERVED = [*ENTITIES, "country/TZA", "country/UGA", "country/ZMB"] +# The sentence 0039 writes for members a range window left out, which the row cannot split into +# off-cadence and out-of-window members; `counted` is `N of M`, or `N` when the trace holds no +# members. +RANGE_MEMBERS_SENTENCE = ( + " {counted} series hold no observations at the group's observation period in the " + "requested years and are left out of the figures." +) MEMBERS = ["dcid:YouthRate", "dcid:AdultRate", "dcid:ElderRate"] MANY_MEMBERS = [f"dcid:Member{index}" for index in range(6)] OMITTED = [f"country/C{index:02}" for index in range(12)] @@ -153,7 +160,15 @@ def warning_refs_cases() -> dict[str, tuple[dict, list[dict[str, object]]]]: ], members=9, ), - [new_coverage(members=MANY_MEMBERS, member_count=9, countries_fallback=True)], + [ + { + "code": "coverage_below_scope", + "severity": "warn", + "message": MATERIALISE_CODE_CATALOGUE["coverage_below_scope"][1] + + RANGE_MEMBERS_SENTENCE.format(counted="6 of 9") + + FALLBACK, + } + ], ), "members_without_trace_count": ( warning_refs_body( diff --git a/tests/dataset_store/test_live.py b/tests/dataset_store/test_live.py index 7477ce1e..3b8e22f8 100644 --- a/tests/dataset_store/test_live.py +++ b/tests/dataset_store/test_live.py @@ -24,7 +24,7 @@ from artifact_store.inventory import LiveArtifactInventory from artifact_store.store import Tier -from dataset_store import ConnectionPool, LiveDatasetStore, connect +from dataset_store import ConnectionPool, LiveDatasetStore, connect, warning_formats from dataset_store.migrate import MIGRATIONS_DIR, apply_pending from dataset_store.snapshot import ( VERSION_FACT_COLUMNS, @@ -37,6 +37,7 @@ from datasets_api.config import DatasetsBuildSettings from datasets_api.doors.definition import DefinitionDeps from datasets_api.doors.pin import pin_artifact +from datasets_api.manifest import warning_model from datasets_api.operation_store import ( _LIVE, _UNTRACKED_LEASE, @@ -56,6 +57,7 @@ SubjectHead, VersionRecord, ) +from materialise.codes import MATERIALISE_CODE_CATALOGUE from series_coverage.index import Estimate from spec_registry import ( IdentityQuarantined, @@ -64,6 +66,14 @@ SpecRegistryStoreError, recompute_identity, ) +from tests.dataset_store._warning_refs import ( + RANGE_MEMBERS_SENTENCE, + counts_since, + format_count_bodies, + format_counts, + warning_refs_body, + warning_refs_cases, +) from tests.datasets_api._legacy_conversion_trace import ( LEGACY_CONVERSION, LEGACY_DEFLATED_CONVERSION, @@ -4072,3 +4082,103 @@ def test_a_kind_enrichment_inserted_after_0033_is_stored_as_party_kind(live_stor } finally: _delete_versions(live_store, [body_only, with_facts]) + + +_WARNING_REFS_MIGRATION = "0039_warning_counted_refs.sql" + + +def test_0039_rewrites_stored_warnings_that_name_refs_into_counted_fields(live_store): + """Each old `coverage_below_scope`, `currency_unmapped_omitted` and + `deflator_missing_omitted` message becomes the counted message and field the writer + produces for the same refs, with a group's `member_count` read from the trace and the + countries-fallback sentence kept. Members a range window left out become one counted + sentence with no field. A member sentence on a trace holding no members counts the + members alone. Warnings already in the new shape, and the stored body, stay as stored, + and a rerun changes nothing.""" + suffix = uuid4().hex + _title, detail = MATERIALISE_CODE_CATALOGUE["coverage_below_scope"] + old_range = ( + f"{detail} No observations were served for dcid:YouthRate, dcid:AdultRate, which hold " + "none at the observation period the other members serve in the requested years." + ) + cases = { + **warning_refs_cases(), + "range_members_without_trace_count": ( + warning_refs_body( + [{"code": "coverage_below_scope", "severity": "warn", "message": old_range}], + members=None, + ), + [ + { + "code": "coverage_below_scope", + "severity": "warn", + "message": detail + RANGE_MEMBERS_SENTENCE.format(counted="2"), + } + ], + ), + } + subjects = {name: f"spec_0039_{name}_{suffix}" for name in cases} + + def _stored() -> dict[str, tuple[list, dict]]: + with live_store.transaction() as tx: + rows = tx.fetch_all( + "SELECT subject_id, warnings, immutable_body FROM datasets.version_record " + "WHERE subject_id = ANY(%s)", + (list(subjects.values()),), + ) + by_subject = {cast(str, row["subject_id"]): row for row in rows} + return { + name: ( + cast(list, by_subject[subject]["warnings"]), + cast(dict, by_subject[subject]["immutable_body"]), + ) + for name, subject in subjects.items() + } + + for name, (body, _expected) in cases.items(): + _insert_as_the_previous_image( + live_store, subject_id=subjects[name], content_version="cv1", body=body + ) + try: + assert _rerun_migration(live_store, _WARNING_REFS_MIGRATION) == [_WARNING_REFS_MIGRATION] + stored = _stored() + assert {name: warnings for name, (warnings, _body) in stored.items()} == { + name: expected for name, (_body, expected) in cases.items() + } + assert {name: body for name, (_warnings, body) in stored.items()} == { + name: body for name, (body, _expected) in cases.items() + } + for warnings, _body in stored.values(): + for warning in warnings: + assert warning_model(warning).model_dump() == warning + + assert _rerun_migration(live_store, _WARNING_REFS_MIGRATION) == [_WARNING_REFS_MIGRATION] + assert _stored() == stored + finally: + _delete_versions(live_store, list(subjects.values())) + + +def test_0039_leaves_no_stored_warning_in_an_old_format(live_store, monkeypatch, capsys): + """Once 0039 has run over a row of every old format, `dataset_store.warning_formats` counts + none of them and no row left to rewrite. A group's `member_count` now matches its trace + on the rows 0039 gave one, and the rows it does not read stay counted as stored.""" + suffix = uuid4().hex + bodies = format_count_bodies() + subjects = [f"spec_0039_formats_{name}_{suffix}" for name in bodies] + before = format_counts(cast(str, _DSN), monkeypatch, capsys) + for subject_id, body in zip(subjects, bodies.values(), strict=True): + _insert_as_the_previous_image( + live_store, subject_id=subject_id, content_version="cv1", body=body + ) + try: + assert _rerun_migration(live_store, _WARNING_REFS_MIGRATION) == [_WARNING_REFS_MIGRATION] + assert counts_since(before, format_counts(cast(str, _DSN), monkeypatch, capsys)) == { + **{name: (0, 0) for name, _condition in warning_formats.FORMATS}, + "currency or deflator, neither named nor counted": (1, 1), + "member_count equals the trace's resolved_stat_vars": (3, 3), + "member_count differs from the trace's resolved_stat_vars": (1, 1), + "member_count beside a trace holding no resolved_stat_vars": (1, 1), + warning_formats.REWRITTEN_ROWS: (None, 0), + } + finally: + _delete_versions(live_store, subjects) diff --git a/tests/dataset_store/test_migrate.py b/tests/dataset_store/test_migrate.py index 0d44b22a..c8a4c8bc 100644 --- a/tests/dataset_store/test_migrate.py +++ b/tests/dataset_store/test_migrate.py @@ -398,6 +398,7 @@ def on_execute(sql: str, params) -> None: "0036_version_record_continuations.sql", "0037_version_record_recency.sql", "0038_version_record_frame_summed.sql", + "0039_warning_counted_refs.sql", ] statements = [sql for sql, _params in conn.executed] assert any("CREATE TABLE datasets.version_record" in sql for sql in statements) diff --git a/tests/platform/test_migrations_immutable.py b/tests/platform/test_migrations_immutable.py index 9cd72fe0..7aa72798 100644 --- a/tests/platform/test_migrations_immutable.py +++ b/tests/platform/test_migrations_immutable.py @@ -95,6 +95,9 @@ "0038_version_record_frame_summed.sql": ( "7236772faed0ec1328386a75b87a8ca85a18374b0e6affabf93d590478129664" ), + "0039_warning_counted_refs.sql": ( + "36ca2cde2a32e9943b1cf2b128db117eda251872671292329f277594352b5fae" + ), } _PRESENTATION_STORE = {