release: apis dev to main

corpus/standard.yaml

@@ -2010,8 +2010,6 @@
   query: German aid to Kenya
   why: The demonym "German" names Germany as the donor, which extraction reads as the place the resolver holds.
   tags: {theme: aid, entity: two_roles, direction: both, time: none, operation: none, door: raw_text}
-  # Known failure: the binder's reply leaves dac2aAidType out, and a slot left out is not read as open, so the total ODA default does not bind.
-  known_failure: {issue: 99, checks: [measure]}
   expect:
     definite:
       measure: {family: dac2a.oda_gross_received}

src/dataset_store/warning_formats.py

@@ -0,0 +1,188 @@
+"""Count the stored warnings that name refs in their message, by old format.
+
+Migration `0039_warning_counted_refs.sql` rewrites these warnings of
+`datasets.version_record.warnings` into counted messages with structured fields. This command
+counts each old format it rewrites, and checks the `member_count` a group's
+`coverage_below_scope` already stores against the length of its trace's
+`resolved_stat_vars`, the count 0039 gives the warnings it rewrites. It reads in a READ ONLY
+transaction, refuses to read when the session is not read-only, and rolls back.
+
+It connects with the migration job's settings (`dataset_store.migrate`), so it runs as an
+execution of that job with its args overridden. The deployed image omits `scripts/`, so this
+command lives in the package. Run it with `python -m dataset_store.warning_formats`.
+"""
+
+from __future__ import annotations
+
+import sys
+from typing import cast
+
+from dataset_store import connection
+from dataset_store.dbapi import DBAPIConnection
+from dataset_store.errors import DatasetStoreConfigError, DatasetStoreError
+
+# Each format and the SQL condition on one stored warning that selects it. A warning reads as
+# `code`, `message`, `body` (the whole warning) and `trace_members`, the length of its row's
+# `trace.resolved_stat_vars`, NULL when the trace holds no such list. The patterns are the
+# sentences 0039 parses.
+_MEMBERS = "which hold none at the observation period the other members serve"
+_UNSERVED = r" No observations were served for [^ ,]+(, [^ ,]+)*( and [0-9]+ more)?\."
+_FALLBACK = "No place was named, so the answer covers the countries with data"
+_NAMED = " [0-9]+ entities: "
+_TRUNCATED = r" [0-9]+ entities: .* \(truncated\)$"
+FORMATS: tuple[tuple[str, str], ...] = (
+    (
+        "coverage_below_scope, unserved entities named",
+        f"code = 'coverage_below_scope' AND message ~ '{_UNSERVED}'",
+    ),
+    (
+        "coverage_below_scope, members named, window not a range",
+        rf"code = 'coverage_below_scope' AND message ~ '{_MEMBERS}\.'",
+    ),
+    (
+        "coverage_below_scope, members named, range window",
+        rf"code = 'coverage_below_scope' AND message ~ '{_MEMBERS} in the requested years\.'",
+    ),
+    (
+        "coverage_below_scope, members named, trace holds no resolved_stat_vars",
+        f"code = 'coverage_below_scope' AND message ~ '{_MEMBERS}' AND trace_members IS NULL",
+    ),
+    (
+        "coverage_below_scope, named refs beside the countries fallback",
+        f"code = 'coverage_below_scope' AND (message ~ '{_UNSERVED}' OR message ~ '{_MEMBERS}') "
+        f"AND strpos(message, '{_FALLBACK}') > 0",
+    ),
+    (
+        "currency_unmapped_omitted, entities named",
+        f"code = 'currency_unmapped_omitted' AND message ~ '{_NAMED}'",
+    ),
+    (
+        "currency_unmapped_omitted, entities named, truncated",
+        f"code = 'currency_unmapped_omitted' AND message ~ '{_TRUNCATED}'",
+    ),
+    (
+        "deflator_missing_omitted, entities named",
+        f"code = 'deflator_missing_omitted' AND message ~ '{_NAMED}'",
+    ),
+    (
+        "deflator_missing_omitted, entities named, truncated",
+        f"code = 'deflator_missing_omitted' AND message ~ '{_TRUNCATED}'",
+    ),
+    (
+        "currency or deflator, neither named nor counted",
+        "code IN ('currency_unmapped_omitted', 'deflator_missing_omitted') "
+        f"AND message !~ '{_NAMED}' "
+        "AND NOT (body ? 'omitted_entities' OR body ? 'omitted_entity_years')",
+    ),
+    (
+        "member_count equals the trace's resolved_stat_vars",
+        "body ? 'member_count' AND (body ->> 'member_count')::integer = trace_members",
+    ),
+    (
+        "member_count differs from the trace's resolved_stat_vars",
+        "body ? 'member_count' AND (body ->> 'member_count')::integer <> trace_members",
+    ),
+    (
+        "member_count beside a trace holding no resolved_stat_vars",
+        "body ? 'member_count' AND trace_members IS NULL",
+    ),
+)
+
+# The formats 0039 rewrites, whose rows it updates.
+_REWRITTEN = (
+    f"code = 'coverage_below_scope' AND (message ~ '{_UNSERVED}' OR message ~ '{_MEMBERS}') "
+    f"OR code IN ('currency_unmapped_omitted', 'deflator_missing_omitted') "
+    f"AND message ~ '{_NAMED}'"
+)
+REWRITTEN_ROWS = "version rows 0039 rewrites"
+
+
+def _query() -> str:
+    """One row: for each of `FORMATS`, its warnings and the version rows holding them, then
+    the version rows holding any warning 0039 rewrites."""
+    counts = ",\n".join(
+        f"count(*) FILTER (WHERE {condition}), "
+        f"count(DISTINCT (subject_id, content_version)) FILTER (WHERE {condition})"
+        for _name, condition in FORMATS
+    )
+    return f"""
+        WITH warning AS (
+            SELECT
+                r.subject_id,
+                r.content_version,
+                w.value ->> 'code' AS code,
+                w.value ->> 'message' AS message,
+                w.value AS body,
+                CASE
+                    WHEN jsonb_typeof(r.trace -> 'resolved_stat_vars') = 'array'
+                        THEN jsonb_array_length(r.trace -> 'resolved_stat_vars')
+                END AS trace_members
+            FROM datasets.version_record AS r
+            CROSS JOIN LATERAL jsonb_array_elements(r.warnings) AS w (value)
+            WHERE jsonb_typeof(r.warnings) = 'array'
+        )
+        SELECT
+            {counts},
+            count(DISTINCT (subject_id, content_version)) FILTER (WHERE {_REWRITTEN})
+        FROM warning
+    """
+
+
+def count_formats(conn: DBAPIConnection) -> list[tuple[str, int | None, int]]:
+    """Each of `FORMATS` with its warnings and the version rows holding them, then
+    `REWRITTEN_ROWS` with its version rows alone. Reads in a READ ONLY transaction and rolls
+    back. Raises `DatasetStoreError` when the session does not report the transaction
+    read-only."""
+    cursor = conn.cursor()
+    try:
+        cursor.execute("SET TRANSACTION READ ONLY")
+        cursor.execute("SHOW transaction_read_only")
+        read_only = cursor.fetchone()
+        if read_only is None or read_only[0] != "on":
+            raise DatasetStoreError(f"transaction_read_only is {read_only!r}, not 'on'")
+        cursor.execute(_query())
+        row = cursor.fetchone()
+    finally:
+        cursor.close()
+        conn.rollback()
+    assert row is not None
+    values = cast("tuple[int, ...]", row)
+    counts: list[tuple[str, int | None, int]] = [
+        (name, values[2 * index], values[2 * index + 1])
+        for index, (name, _condition) in enumerate(FORMATS)
+    ]
+    return [*counts, (REWRITTEN_ROWS, None, values[-1])]
+
+
+def main() -> int:
+    """Run `python -m dataset_store.warning_formats`. Prints one line per format; a
+    configuration or store error exits 1."""
+    try:
+        settings = connection.ConnectionSettings.from_env(
+            user_var="DATASET_STORE_MIGRATOR_DB_USER",
+            password_var="DATASET_STORE_MIGRATOR_DB_PASSWORD",
+            enable_iam_auth=False,
+        )
+    except DatasetStoreConfigError as exc:
+        print(f"ERROR: {exc}", file=sys.stderr)
+        return 1
+
+    connector = connection.build_connector() if settings.instance_connection_name else None
+    conn = settings.connect_callable(connector=connector)()
+    try:
+        counts = count_formats(conn)
+    except DatasetStoreError as exc:
+        print(f"ERROR: {exc}", file=sys.stderr)
+        return 1
+    finally:
+        conn.close()
+    for name, warnings, rows in counts:
+        if warnings is None:
+            print(f"{name}: {rows}")
+        else:
+            print(f"{name}: {warnings} warnings in {rows} version rows")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/datasets_api/operation_store.py

@@ -96,7 +96,7 @@
 
 from pydantic import ValidationError
 
-from dataset_store import DatasetStore
+from dataset_store import DatasetStore, Transaction
 from datasets_api.backpressure import Backpressure
 from datasets_api.config import DatasetsBuildSettings
 from datasets_api.operations.lease import LEASE_SECONDS
@@ -174,6 +174,11 @@
 # Serialises the backlog count and insert across transactions.
 _BACKLOG_LOCK_KEY = 0x4446_4F5F_4F50_424C  # "DFO_OPBL" packed into an int
 
+# Passes `create_or_get` makes under the backlog lock. A pass whose insert or reset writes no row
+# saw a concurrent commit change the row after its read, so the next pass decides again on the
+# row as it now stands. Each extra pass needs another such commit inside the same transaction.
+_CREATE_PASSES = 3
+
 
 def utc_now() -> datetime:
     return datetime.now(UTC)
@@ -606,7 +611,6 @@ def create_or_get(
             options=options,
         )
         now = self._now()
-        backlog_max = self._backlog_max
 
         def _run() -> Mapping[str, object] | None:
             with self._store.transaction() as tx:
@@ -617,70 +621,91 @@ def _run() -> Mapping[str, object] | None:
                     return existing
                 # Lock before counting or inserting a new row or resetting one.
                 tx.execute("SELECT pg_advisory_xact_lock(%s)", (_BACKLOG_LOCK_KEY,))
-                # Re-read under the lock. This statement's fresh snapshot controls the mutation.
-                existing = tx.fetch_one(
-                    f"SELECT {_COLUMNS} FROM datasets.operation WHERE id = %s", (id,)
-                )
-                if existing is not None and not _resettable(existing, now, deadline=self._deadline):
-                    return existing
-                # The row being reset keeps its own place, so only the others count.
-                backlog_row = tx.fetch_one(
-                    f"SELECT count(*) AS n FROM datasets.operation WHERE id <> %s AND {_LIVE}",
-                    (id, *_abandoned_params(now, deadline=self._deadline)),
-                )
-                in_flight = cast(int, backlog_row["n"]) if backlog_row is not None else 0
-                if in_flight >= backlog_max:
-                    raise OperationBacklogFull(in_flight=in_flight, limit=backlog_max)
-                if existing is None:
-                    tx.execute(
-                        "INSERT INTO datasets.operation "
-                        "(id, request_identity, options, coverage_stamp, state, estimate_rows, "
-                        "estimate_requests, rows_written, attempt_count, created_at, updated_at, "
-                        "submitted_at, origin_catalogue_build_id) "
-                        "VALUES (%s, %s, %s, %s, 'pending', %s, %s, 0, 0, %s, %s, %s, %s) "
-                        "ON CONFLICT (id) DO NOTHING",
-                        (
-                            id,
-                            dict(request_identity),
-                            dict(options),
-                            coverage_stamp,
-                            estimate.rows,
-                            estimate.requests,
-                            now,
-                            now,
-                            now,
-                            origin_catalogue_build_id,
-                        ),
+                for _ in range(_CREATE_PASSES):
+                    row = self._create_or_reset_locked(
+                        tx,
+                        id=id,
+                        request_identity=request_identity,
+                        options=options,
+                        coverage_stamp=coverage_stamp,
+                        origin_catalogue_build_id=origin_catalogue_build_id,
+                        estimate=estimate,
+                        now=now,
                     )
-                else:
-                    # Reset only the resettable row seen under the backlog lock.
-                    tx.execute(
-                        "UPDATE datasets.operation SET state = 'pending', "
-                        "claimed_at = NULL, lease_seconds = NULL, final_delivery = NULL, "
-                        "refusal_code = NULL, refusal = NULL, retryable = NULL, rows_written = 0, "
-                        "submission_count = submission_count + 1, dispatched_at = NULL, "
-                        "submitted_at = %s, updated_at = %s "
-                        f"WHERE id = %s AND {_RESETTABLE}",
-                        (
-                            now,
-                            now,
-                            id,
-                            now,
-                            now - _UNTRACKED_LEASE,
-                            now - self._deadline,
-                        ),
-                    )
-                return tx.fetch_one(
-                    f"SELECT {_COLUMNS} FROM datasets.operation WHERE id = %s", (id,)
-                )
+                    if row is not None:
+                        return row
+                return None
 
         row = self._wrap("creating", id, _run)
         if row is None:
             raise DatasetsStoreUnavailable(
-                f"creating operation {id!r} failed: insert-then-select returned no row"
+                f"creating operation {id!r} failed: the row changed under each of "
+                f"{_CREATE_PASSES} attempts to insert or reset it"
             )
         return _operation_from_row(row)
 
+    def _create_or_reset_locked(
+        self,
+        tx: Transaction,
+        *,
+        id: str,
+        request_identity: Mapping[str, object],
+        options: Mapping[str, object],
+        coverage_stamp: str,
+        origin_catalogue_build_id: str,
+        estimate: Estimate,
+        now: datetime,
+    ) -> Mapping[str, object] | None:
+        """One pass of `create_or_get` under the backlog lock.
+
+        Returns the row as read when it is not resettable, else the row the insert or reset
+        wrote. Returns `None` when a commit since the read left the insert or reset nothing to
+        write, which leaves the decision to the next pass.
+        """
+        # Each statement takes a fresh snapshot, so re-read under the lock.
+        existing = tx.fetch_one(f"SELECT {_COLUMNS} FROM datasets.operation WHERE id = %s", (id,))
+        if existing is not None and not _resettable(existing, now, deadline=self._deadline):
+            return existing
+        # The row being reset keeps its own place, so only the others count.
+        backlog_row = tx.fetch_one(
+            f"SELECT count(*) AS n FROM datasets.operation WHERE id <> %s AND {_LIVE}",
+            (id, *_abandoned_params(now, deadline=self._deadline)),
+        )
+        in_flight = cast(int, backlog_row["n"]) if backlog_row is not None else 0
+        if in_flight >= self._backlog_max:
+            raise OperationBacklogFull(in_flight=in_flight, limit=self._backlog_max)
+        if existing is None:
+            return tx.fetch_one(
+                "INSERT INTO datasets.operation "
+                "(id, request_identity, options, coverage_stamp, state, estimate_rows, "
+                "estimate_requests, rows_written, attempt_count, created_at, updated_at, "
+                "submitted_at, origin_catalogue_build_id) "
+                "VALUES (%s, %s, %s, %s, 'pending', %s, %s, 0, 0, %s, %s, %s, %s) "
+                f"ON CONFLICT (id) DO NOTHING RETURNING {_COLUMNS}",
+                (
+                    id,
+                    dict(request_identity),
+                    dict(options),
+                    coverage_stamp,
+                    estimate.rows,
+                    estimate.requests,
+                    now,
+                    now,
+                    now,
+                    origin_catalogue_build_id,
+                ),
+            )
+        # `_RESETTABLE` re-checks the row as it stands when the update runs.
+        return tx.fetch_one(
+            "UPDATE datasets.operation SET state = 'pending', "
+            "claimed_at = NULL, lease_seconds = NULL, final_delivery = NULL, "
+            "refusal_code = NULL, refusal = NULL, retryable = NULL, rows_written = 0, "
+            "submission_count = submission_count + 1, dispatched_at = NULL, "
+            "submitted_at = %s, updated_at = %s "
+            f"WHERE id = %s AND {_RESETTABLE} RETURNING {_COLUMNS}",
+            (now, now, id, now, now - _UNTRACKED_LEASE, now - self._deadline),
+        )
+
     def mark_dispatched(self, id: str, *, submission_count: int) -> Operation | None:
         now = self._now()
         return self._update_one(