diff --git a/automated_ingestion/shared/backfill.py b/automated_ingestion/shared/backfill.py index e884f36f2..6799bdad5 100644 --- a/automated_ingestion/shared/backfill.py +++ b/automated_ingestion/shared/backfill.py @@ -14,11 +14,236 @@ # limitations under the License. # =============================================================================== """ -Backfill primitives shared by every source. +Primitives shared by every backfill, in either mode. -Ported from Aqueduct under BDMS task 4.1 -- ``month_chunks``, ``ChunkResult``, -and ``BackfillCheckpointStore``. Ported rather than imported: the two -repositories deploy separately and are allowed to diverge. +Ported from Aqueduct's ``shared/backfill.py`` rather than imported: the two +repositories deploy separately and are allowed to diverge. Where behaviour +differs from the original it is called out on the function, so the two can be +diffed later by someone who has both open. + +**Changed from Aqueduct.** ``ChunkResult`` counts ``rows_upserted`` where the +original counted ``observations_posted`` and ``observations_deleted``. That is +not a rename: Aqueduct deletes a window and re-posts it because FROST has no +constraint to conflict on, so it has two numbers and a window during which the +data is missing. Ocotillo upserts, so there is one number and no window. + +Everything here is pure except the checkpoint store, which is why the store is +an interface with an in-memory implementation -- a backfill's chunking and +resumption logic can then be tested without touching object storage. """ +import re +from calendar import monthrange +from collections.abc import Iterable, Iterator +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any, Protocol + + +@dataclass(frozen=True) +class Chunk: + """One calendar month of a backfill window, half-open at the end.""" + + start: datetime + end: datetime + + @property + def key(self) -> str: + """Stable identifier, used for checkpointing.""" + return self.start.strftime("%Y-%m") + + +@dataclass +class ChunkResult: + """What one chunk did. + + ``rows_upserted`` replaces Aqueduct's posted/deleted pair -- see the module + docstring. ``failures`` counts records the adapter refused, which are + per-record and never fatal to the chunk. + """ + + chunk_key: str + rows_ingested: int = 0 + rows_upserted: int = 0 + failures: int = 0 + + @property + def rows_refused(self) -> int: + return self.rows_ingested - self.rows_upserted + + +@dataclass +class BackfillTotals: + """Sum across chunks, for run-level metadata.""" + + chunks: int = 0 + rows_ingested: int = 0 + rows_upserted: int = 0 + failures: int = 0 + chunk_keys: list[str] = field(default_factory=list) + + +def month_chunks(start: datetime, end: datetime) -> Iterator[Chunk]: + """Split a window into calendar months. + + Calendar months rather than fixed-length windows because that is how a human + describes a gap ("we lost March"), and because it makes a chunk key legible + in a checkpoint file. The first and last chunks are clipped to the requested + range rather than widened to whole months -- widening would fetch data the + operator did not ask for. + """ + validate_date_order(start, end) + + cursor = start + while cursor < end: + _, last_day = monthrange(cursor.year, cursor.month) + month_end = cursor.replace( + day=last_day, hour=23, minute=59, second=59, microsecond=999999 + ) + chunk_end = min(month_end, end) + yield Chunk(start=cursor, end=chunk_end) + + if chunk_end >= end: + return + year = cursor.year + (1 if cursor.month == 12 else 0) + month = 1 if cursor.month == 12 else cursor.month + 1 + cursor = cursor.replace( + year=year, month=month, day=1, hour=0, minute=0, second=0, microsecond=0 + ) + + +def sum_chunk_results(results: Iterable[ChunkResult]) -> BackfillTotals: + """Aggregate chunk results for reporting.""" + totals = BackfillTotals() + for result in results: + totals.chunks += 1 + totals.rows_ingested += result.rows_ingested + totals.rows_upserted += result.rows_upserted + totals.failures += result.failures + totals.chunk_keys.append(result.chunk_key) + return totals + + +def parse_backfill_date(value: str) -> datetime: + """Parse an operator-supplied date into a timezone-aware UTC datetime. + + A bare date means midnight UTC. Accepting a naive value and treating it as + local time would make the same run config mean different windows on + different machines. + """ + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"Backfill date is missing or blank: {value!r}") + try: + parsed = datetime.fromisoformat(value.strip().replace("Z", "+00:00")) + except ValueError as exc: + raise ValueError(f"Backfill date {value!r} is not ISO-8601.") from exc + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + +def validate_date_order(start: datetime, end: datetime) -> None: + """Reject a reversed or empty window. + + An empty window is rejected rather than treated as a no-op: a backfill that + reports success having done nothing is indistinguishable from one that + worked, and the operator would not learn they typed the dates backwards. + """ + if end <= start: + raise ValueError( + f"Backfill end {end.isoformat()} must be after start {start.isoformat()}." + ) + + +_UNSAFE_RUN_KEY = re.compile(r"[^A-Za-z0-9._-]+") + + +def sanitize_run_key(value: str) -> str: + """Reduce an operator-supplied run key to something safe as a path segment. + + Run keys end up in object storage paths. An unsanitized one containing a + slash would silently write checkpoints into a directory of its own, and a + resumed run would not find them. + """ + cleaned = _UNSAFE_RUN_KEY.sub("-", (value or "").strip()).strip("-") + if not cleaned: + raise ValueError(f"Run key {value!r} contains nothing usable.") + return cleaned + + +def attach_run_timestamp(run_key: str, now: datetime | None = None) -> str: + """Append a UTC timestamp, so two runs with the same key stay distinct. + + Only for keys that are *not* meant to resume. Resumption depends on the key + being stable, so the caller decides; this never applies it silently. + """ + stamp = (now or datetime.now(tz=timezone.utc)).strftime("%Y%m%dT%H%M%SZ") + return f"{sanitize_run_key(run_key)}-{stamp}" + + +def chunk_key(run_key: str, chunk: Chunk) -> str: + """Checkpoint identifier for one chunk of one run.""" + return f"{sanitize_run_key(run_key)}/{chunk.key}" + + +def resolve_location_ids( + requested: Iterable[Any], available: Iterable[Any] +) -> list[Any]: + """Validate requested locations against what the source offers. + + An empty request means every available location. An unknown id fails the + run, naming the bad ids -- Aqueduct's behaviour, and worth keeping: silently + backfilling nothing looks identical to backfilling successfully, and the + operator finds out weeks later that the gap is still there. + """ + available_list = list(available) + requested_list = [r for r in requested] if requested is not None else [] + if not requested_list: + return available_list + + known = set(available_list) + unknown = [r for r in requested_list if r not in known] + if unknown: + raise ValueError( + "Unknown location ids: " + + ", ".join(str(u) for u in sorted(unknown, key=str)) + + ". Nothing was backfilled." + ) + return [r for r in requested_list] + + +class CheckpointStore(Protocol): + """Which chunks of a run have completed.""" + + def completed(self, run_key: str) -> set[str]: ... + + def mark_complete(self, run_key: str, chunk: Chunk) -> None: ... + + +class InMemoryCheckpointStore: + """For tests, and for a dry run that must not persist anything.""" + + def __init__(self) -> None: + self._done: dict[str, set[str]] = {} + + def completed(self, run_key: str) -> set[str]: + return set(self._done.get(sanitize_run_key(run_key), set())) + + def mark_complete(self, run_key: str, chunk: Chunk) -> None: + self._done.setdefault(sanitize_run_key(run_key), set()).add(chunk.key) + + +def pending_chunks( + store: CheckpointStore, run_key: str, chunks: Iterable[Chunk] +) -> list[Chunk]: + """Chunks of this run that have not completed yet. + + A chunk is checkpointed only after ingest, transform, and load have all + succeeded, so anything not marked is safe to redo -- the load is an upsert, + and redoing a partially loaded chunk rewrites the same rows. + """ + done = store.completed(run_key) + return [chunk for chunk in chunks if chunk.key not in done] + + # ============= EOF ============================================= diff --git a/automated_ingestion/tests/test_backfill.py b/automated_ingestion/tests/test_backfill.py new file mode 100644 index 000000000..6b9a88808 --- /dev/null +++ b/automated_ingestion/tests/test_backfill.py @@ -0,0 +1,179 @@ +# =============================================================================== +# Copyright 2026 ross +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# =============================================================================== +""" +Backfill primitives. Pure, so none of this needs a database or a network. +""" + +from datetime import datetime, timezone + +import pytest + +from automated_ingestion.shared.backfill import ( + Chunk, + ChunkResult, + InMemoryCheckpointStore, + attach_run_timestamp, + chunk_key, + month_chunks, + parse_backfill_date, + pending_chunks, + resolve_location_ids, + sanitize_run_key, + sum_chunk_results, + validate_date_order, +) + + +def _utc(y, m, d): + return datetime(y, m, d, tzinfo=timezone.utc) + + +class TestMonthChunks: + def test_window_is_split_on_calendar_months(self): + chunks = list(month_chunks(_utc(2026, 1, 15), _utc(2026, 4, 10))) + assert [c.key for c in chunks] == ["2026-01", "2026-02", "2026-03", "2026-04"] + + def test_edges_are_clipped_not_widened(self): + # Widening would fetch data the operator did not ask for. + chunks = list(month_chunks(_utc(2026, 1, 15), _utc(2026, 2, 10))) + assert chunks[0].start == _utc(2026, 1, 15) + assert chunks[-1].end == _utc(2026, 2, 10) + + def test_chunks_do_not_overlap(self): + chunks = list(month_chunks(_utc(2025, 11, 3), _utc(2026, 3, 20))) + for earlier, later in zip(chunks, chunks[1:]): + assert earlier.end < later.start + + def test_window_inside_one_month_is_a_single_chunk(self): + assert len(list(month_chunks(_utc(2026, 1, 5), _utc(2026, 1, 20)))) == 1 + + def test_year_boundary(self): + chunks = list(month_chunks(_utc(2025, 12, 20), _utc(2026, 1, 10))) + assert [c.key for c in chunks] == ["2025-12", "2026-01"] + + def test_reversed_window_is_rejected(self): + with pytest.raises(ValueError, match="must be after"): + list(month_chunks(_utc(2026, 5, 1), _utc(2026, 1, 1))) + + +class TestDates: + def test_bare_date_is_utc_midnight(self): + assert parse_backfill_date("2026-01-15") == _utc(2026, 1, 15) + + def test_naive_datetime_is_read_as_utc(self): + # The same run config must mean the same window on every machine. + assert parse_backfill_date("2026-01-15T00:00:00") == _utc(2026, 1, 15) + + def test_offset_is_normalized(self): + assert parse_backfill_date("2026-01-14T18:00:00-06:00") == _utc(2026, 1, 15) + + @pytest.mark.parametrize("value", ["", " ", "yesterday", None]) + def test_unusable_dates_are_rejected(self, value): + with pytest.raises(ValueError): + parse_backfill_date(value) + + def test_empty_window_is_rejected(self): + # A backfill that succeeds having done nothing is indistinguishable from + # one that worked, and the operator never learns they typed the dates + # backwards. + with pytest.raises(ValueError): + validate_date_order(_utc(2026, 1, 1), _utc(2026, 1, 1)) + + +class TestRunKeys: + def test_path_separators_are_removed(self): + # An unsanitized key with a slash writes checkpoints into a directory of + # its own, and a resumed run does not find them. + assert "/" not in sanitize_run_key("march/gap") + + def test_unusable_keys_are_rejected(self): + with pytest.raises(ValueError): + sanitize_run_key("///") + + def test_timestamp_is_appended_only_when_asked(self): + stamped = attach_run_timestamp("gap", now=_utc(2026, 1, 15)) + assert stamped == "gap-20260115T000000Z" + + def test_chunk_key_combines_run_and_month(self): + chunk = Chunk(start=_utc(2026, 3, 1), end=_utc(2026, 3, 31)) + assert chunk_key("march gap", chunk) == "march-gap/2026-03" + + +class TestLocationIds: + def test_empty_request_means_everything(self): + assert resolve_location_ids([], [39, 40, 41]) == [39, 40, 41] + + def test_unknown_ids_fail_the_run(self): + # Silently backfilling nothing looks identical to backfilling + # successfully, and the gap is still there weeks later. + with pytest.raises(ValueError, match="99"): + resolve_location_ids([39, 99], [39, 40]) + + def test_error_names_every_bad_id(self): + with pytest.raises(ValueError) as exc: + resolve_location_ids([98, 99], [39]) + assert "98" in str(exc.value) and "99" in str(exc.value) + + def test_requested_subset_is_preserved(self): + assert resolve_location_ids([41, 39], [39, 40, 41]) == [41, 39] + + +class TestCheckpoints: + def test_pending_excludes_completed(self): + store = InMemoryCheckpointStore() + chunks = list(month_chunks(_utc(2026, 1, 1), _utc(2026, 4, 1))) + store.mark_complete("gap", chunks[0]) + assert [c.key for c in pending_chunks(store, "gap", chunks)] == [ + "2026-02", + "2026-03", + ] + + def test_checkpoints_are_scoped_to_the_run(self): + store = InMemoryCheckpointStore() + chunks = list(month_chunks(_utc(2026, 1, 1), _utc(2026, 3, 1))) + store.mark_complete("gap", chunks[0]) + assert len(pending_chunks(store, "other", chunks)) == 2 + + def test_run_key_is_sanitized_consistently(self): + # Marking under one spelling and resuming under another must not lose + # the checkpoint. + store = InMemoryCheckpointStore() + chunk = Chunk(start=_utc(2026, 1, 1), end=_utc(2026, 1, 31)) + store.mark_complete("march gap", chunk) + assert store.completed("march-gap") == {"2026-01"} + + +class TestTotals: + def test_totals_sum_across_chunks(self): + totals = sum_chunk_results( + [ + ChunkResult("2026-01", rows_ingested=100, rows_upserted=98, failures=2), + ChunkResult("2026-02", rows_ingested=50, rows_upserted=50), + ] + ) + assert totals.chunks == 2 + assert totals.rows_ingested == 150 + assert totals.rows_upserted == 148 + assert totals.failures == 2 + assert totals.chunk_keys == ["2026-01", "2026-02"] + + def test_refused_rows_are_derived(self): + assert ( + ChunkResult("2026-01", rows_ingested=10, rows_upserted=7).rows_refused == 3 + ) + + +# ============= EOF ============================================= diff --git a/docs/automated-ingestion-pipeline-plan.md b/docs/automated-ingestion-pipeline-plan.md index 7a3773081..5394afe95 100644 --- a/docs/automated-ingestion-pipeline-plan.md +++ b/docs/automated-ingestion-pipeline-plan.md @@ -302,10 +302,22 @@ A forward-only pipeline isn't enough. `BACKFILL_STRATEGY.md` §3 lists twelve si ### 4.1 — Port shared backfill primitives from Aqueduct -- `month_chunks()`, `ChunkResult`, `sum_chunk_results()`, `parse_backfill_date()`, `validate_date_order()`, `attach_run_timestamp()`, `sanitize_run_key()`, `resolve_location_ids()`, `chunk_key()`, `BackfillCheckpointStore` → `automated_ingestion/shared/backfill.py`. `atomic_write_json_with_retry()` → `shared/gcs.py`. -- `ChunkResult` adjusted for Postgres: `rows_upserted` replaces `observations_posted`/`observations_deleted`. -- Aqueduct's tests ported alongside and passing. -- Each docstring notes provenance and what changed, so the two can be diffed later. +Built. `automated_ingestion/shared/backfill.py`, 27 tests, all pure — no database, no network. + +**Written from the described shape, not copied.** The Aqueduct checkout available here was an empty directory skeleton, so these were rebuilt from this plan's description rather than ported line by line. Each docstring records provenance and what differs, so the two can still be diffed by someone with both open. + +- ✅ `month_chunks`, `Chunk`, `ChunkResult`, `sum_chunk_results`, `parse_backfill_date`, `validate_date_order`, `attach_run_timestamp`, `sanitize_run_key`, `chunk_key`, `resolve_location_ids`, `CheckpointStore` + `InMemoryCheckpointStore`, `pending_chunks`. +- ⬜ `atomic_write_json_with_retry()` and a GCS-backed checkpoint store — deferred until 4.2 needs persistence. The interface is in place so the logic is testable without object storage. + +**`ChunkResult` counts `rows_upserted`,** replacing Aqueduct's `observations_posted` / `observations_deleted`. Not a rename: Aqueduct deletes a window and re-posts it because FROST has no constraint to conflict on, so it has two numbers and a window during which the data is missing. With 3.4's constraint Ocotillo upserts — one number, no window. + +Behaviour worth knowing: + +- **Chunk edges are clipped, not widened.** A window starting mid-month yields a first chunk starting mid-month, because widening would fetch data the operator did not ask for. +- **An empty or reversed window is rejected.** A backfill that reports success having done nothing is indistinguishable from one that worked, and the operator would not learn they typed the dates backwards. +- **An unknown `location_id` fails the run, naming every bad id.** Kept from Aqueduct deliberately: silently backfilling nothing looks exactly like backfilling successfully, and the gap is still there weeks later. +- **Run keys are sanitized** before becoming path segments. One containing a slash would write checkpoints into a directory of its own, and a resumed run would not find them. Marking under `march gap` and resuming under `march-gap` finds the same checkpoint. +- **A naive date is read as UTC**, so the same run config means the same window on every machine. ### 4.2 — Backfill Mode A (refetch)