Shelf Analytics

Backfilling Compliance History After a Schema Change

This walkthrough sits under Time-Series Compliance Drift Analysis and solves one precise task: after the compliance payload schema or the scoring logic changes, recompute the historical aggregates so the drift charts stay honest — without corrupting the live series or double-counting a single capture. A schema bump is unavoidable in a maturing pipeline: you add price_tag_mismatch_count, you re-weight misplaced_sku_list against out_of_stock_flags, or a registry_revision rolls the planogram under a store reset. The moment that lands, every historical compliance_percentage was computed under the old rules, so a 2026-03 figure and a 2026-07 figure are no longer measured with the same ruler, and the seasonal drift the parent cluster hunts becomes an artifact of your own release. The fix is a backfill that reads immutable raw payloads, recomputes aggregates in registry_revision-aware batches, and lands them through an idempotent upsert keyed on the natural key — so a re-run converges instead of accumulating. This page builds that backfill step by step, and each step is independently verifiable.

Idempotent compliance backfill flow: raw payloads to batched recompute to idempotent upsert to parity check to cutover Immutable raw payloads keyed by fixture_id, capture_timestamp, and registry_revision are read into a recompute stage that processes them in batches grouped by registry_revision. Each recomputed aggregate row lands through an idempotent upsert keyed on the natural key, so re-running the backfill overwrites the same row rather than inserting a duplicate. Rows are written into a versioned shadow aggregate table that sits beside the live series and never touches it during the run. A parity gate then compares old live aggregates against the new shadow rows within a tolerance; only when parity passes does an atomic cutover swap the shadow table into place, at which point dashboards read the recomputed history. If parity fails the shadow table is discarded and the live series is untouched. immutable raw payloads keyed on fixture_id + capture_timestamp + registry_revision batched recompute grouped by registry_revision new scoring logic idempotent upsert on natural key re-run overwrites, never duplicates into versioned shadow table parity check old live vs. new shadow row counts + spot deltas atomic cutover swap shadow into live series parity fails → discard shadow, live untouched Idempotent backfill: raw payloads never mutate; the live series only changes at cutover Re-running any stage converges on the same aggregates because every write is keyed on the natural key.

Prerequisites & Context

Before running a backfill, confirm these foundations are in place. A backfill is only as safe as the immutability guarantees underneath it — if raw payloads can be edited in place, no re-run is reproducible.

  • Runtime: Python 3.11+ with a database driver (psycopg for Postgres in the examples) and pydantic on the batch host. No GPU; this is pure aggregation over stored records.
  • Immutable raw payloads: every compliance capture is stored once, write-once, keyed on the natural key fixture_id + capture_timestamp + registry_revision. The backfill reads these and never mutates them — they are the source of truth the recompute replays against. If your raw store allows in-place edits, freeze it first.
  • A versioned aggregate store: the derived table the dashboards read must carry a schema_version (or scoring_version) column, so old and new aggregates can coexist during the run. This is the same versioning discipline that Versioning Compliance Payloads Across Store Resets applies upstream — a schema bump there is exactly what triggers a backfill here.
  • A defined backfill window: the [start, end) capture-timestamp range and the set of registry_revision values in scope. Recomputing all of history when only one revision changed wastes hours and risks the live series for no gain.
  • A pinned scoring function: the new scoring logic, version-tagged, so a recomputed row records which algorithm produced it. Without this you cannot tell a backfilled row from a live one, and parity checks become guesswork.

A note on terms: idempotent here means running the backfill once, twice, or ten times lands the same rows — because every write targets a row identified by the natural key, a repeat overwrites rather than appends.

Step 1 — Snapshot the Live Series and Define the Window

Never recompute in place. Before touching anything, snapshot the current aggregates for the window and materialize the recomputed rows into a shadow table that carries the new schema_version. The live series keeps serving dashboards untouched until the parity gate passes. The snapshot is also your rollback: if the cutover misbehaves, you restore it verbatim.

Start by pinning the window and the revisions in scope, then copy the live rows aside.

from __future__ import annotations

import datetime as dt
from dataclasses import dataclass


@dataclass(frozen=True)
class BackfillWindow:
    """The bounded scope of one backfill run."""
    start: dt.datetime          # inclusive, UTC-aware
    end: dt.datetime            # exclusive, UTC-aware
    registry_revisions: tuple[str, ...]
    new_schema_version: int

    def __post_init__(self) -> None:
        if self.start.tzinfo is None or self.end.tzinfo is None:
            raise ValueError("window bounds must be timezone-aware (UTC)")
        if self.start >= self.end:
            raise ValueError("start must be strictly before end")
        if not self.registry_revisions:
            raise ValueError("at least one registry_revision must be in scope")


def snapshot_live(cur, window: BackfillWindow) -> int:
    """Copy live aggregates in scope into a rollback table; return row count."""
    cur.execute(
        """
        CREATE TABLE IF NOT EXISTS compliance_agg_snapshot
            (LIKE compliance_agg INCLUDING ALL);
        INSERT INTO compliance_agg_snapshot
        SELECT * FROM compliance_agg
        WHERE capture_timestamp >= %(start)s
          AND capture_timestamp <  %(end)s
          AND registry_revision = ANY(%(revs)s)
        ON CONFLICT DO NOTHING;
        """,
        {"start": window.start, "end": window.end,
         "revs": list(window.registry_revisions)},
    )
    return cur.rowcount

Keeping the window timezone-aware and exclusive on the upper bound ([start, end)) is what stops the boundary double-count described in Troubleshooting — an inclusive end re-scores the midnight capture in two adjacent runs.

Step 2 — Land Rows Through an Idempotent Upsert on the Natural Key

The heart of a safe backfill is that every write is an upsert keyed on the natural key, so a re-run converges rather than accumulates. Model the aggregate row explicitly, then write it with ON CONFLICT ... DO UPDATE on the natural-key unique constraint. If the batch host dies mid-run and you restart, the already-written rows are simply overwritten with identical values — no duplicates, no drift.

from pydantic import BaseModel, Field


class ComplianceAggRow(BaseModel):
    """One recomputed aggregate, uniquely identified by the natural key."""
    fixture_id: str
    capture_timestamp: dt.datetime
    registry_revision: str
    schema_version: int
    compliance_percentage: float = Field(ge=0.0, le=100.0)
    out_of_stock_flags: int = Field(ge=0)
    misplaced_sku_count: int = Field(ge=0)
    price_tag_mismatch_count: int = Field(ge=0)


UPSERT_SQL = """
INSERT INTO compliance_agg_shadow
    (fixture_id, capture_timestamp, registry_revision, schema_version,
     compliance_percentage, out_of_stock_flags,
     misplaced_sku_count, price_tag_mismatch_count)
VALUES (%(fixture_id)s, %(capture_timestamp)s, %(registry_revision)s,
        %(schema_version)s, %(compliance_percentage)s, %(out_of_stock_flags)s,
        %(misplaced_sku_count)s, %(price_tag_mismatch_count)s)
ON CONFLICT (fixture_id, capture_timestamp, registry_revision, schema_version)
DO UPDATE SET
    compliance_percentage    = EXCLUDED.compliance_percentage,
    out_of_stock_flags       = EXCLUDED.out_of_stock_flags,
    misplaced_sku_count      = EXCLUDED.misplaced_sku_count,
    price_tag_mismatch_count = EXCLUDED.price_tag_mismatch_count;
"""


def upsert_batch(cur, rows: list[ComplianceAggRow]) -> int:
    """Idempotently upsert a batch; safe to replay verbatim."""
    if not rows:
        return 0
    cur.executemany(UPSERT_SQL, [r.model_dump() for r in rows])
    return len(rows)

The schema_version is part of the conflict target on purpose: it lets the new aggregates land beside the old ones in the shadow table without collision, which is what makes the parity comparison in Step 4 a straight join rather than a diff of two tables.

Step 3 — Recompute in registry_revision-Aware Batches

Recompute in bounded batches grouped by registry_revision, never in one pass over all of history. Grouping by revision matters because the scoring rules and the planogram geometry are constant within a revision but change across one — batching along that seam keeps each batch internally consistent and lets you parallelize or resume per revision. Bounding the batch size is what keeps memory flat on a multi-year window; you stream raw payloads in chunks and commit per batch so a failure loses one batch, not the whole run.

from collections.abc import Iterator


def iter_raw_batches(
    cur, window: BackfillWindow, batch_size: int = 5_000,
) -> Iterator[list[dict]]:
    """Yield raw payloads in bounded, revision-ordered batches (keyset paging)."""
    for revision in window.registry_revisions:
        last_key: tuple[str, dt.datetime] | None = None
        while True:
            cur.execute(
                """
                SELECT fixture_id, capture_timestamp, registry_revision, payload
                FROM compliance_raw
                WHERE registry_revision = %(rev)s
                  AND capture_timestamp >= %(start)s
                  AND capture_timestamp <  %(end)s
                  AND (%(last_ts)s IS NULL OR
                       (fixture_id, capture_timestamp) > (%(last_fx)s, %(last_ts)s))
                ORDER BY fixture_id, capture_timestamp
                LIMIT %(limit)s
                """,
                {"rev": revision, "start": window.start, "end": window.end,
                 "last_fx": last_key[0] if last_key else None,
                 "last_ts": last_key[1] if last_key else None,
                 "limit": batch_size},
            )
            batch = cur.fetchall()
            if not batch:
                break
            yield batch
            last = batch[-1]
            last_key = (last["fixture_id"], last["capture_timestamp"])


def run_backfill(conn, window: BackfillWindow, score_fn) -> int:
    """Recompute and upsert the whole window in bounded batches."""
    total = 0
    for batch in iter_raw_batches(conn.cursor(), window):
        rows: list[ComplianceAggRow] = []
        for raw in batch:
            try:
                metrics = score_fn(raw["payload"])           # new scoring logic
            except (KeyError, ValueError) as exc:
                # a malformed historical payload must not abort the run
                _log_skip(raw["fixture_id"], raw["capture_timestamp"], exc)
                continue
            rows.append(ComplianceAggRow(
                fixture_id=raw["fixture_id"],
                capture_timestamp=raw["capture_timestamp"],
                registry_revision=raw["registry_revision"],
                schema_version=window.new_schema_version,
                **metrics,
            ))
        with conn.cursor() as write_cur:
            total += upsert_batch(write_cur, rows)
        conn.commit()          # commit per batch so a crash resumes, not restarts
    return total

Keyset paging (rather than OFFSET) keeps each page cheap on a large table, and committing per batch means a crashed run resumes from the last committed batch — combined with the idempotent upsert, even re-reading a partially processed batch is harmless. The sibling walkthrough Detecting Seasonal Compliance Drift with Python consumes exactly this recomputed series, so a clean backfill is what makes its seasonal decomposition trustworthy.

Step 4 — Validate Parity Old-vs-New, Then Cut Over

Before the shadow table becomes the live series, prove it. Parity has three cheap, decisive checks: row counts must match the snapshot (every live row got a recomputed counterpart), coverage must be complete (no fixture_id × timestamp in scope is missing), and a spot delta on aggregate movement must stay within an expected band — a schema change should move scores, but a 40-point swing on every fixture signals a scoring bug, not a legitimate re-weight. Only when parity passes do you swap the shadow table into place atomically.

@dataclass(frozen=True)
class ParityReport:
    live_rows: int
    shadow_rows: int
    missing_keys: int
    max_abs_delta: float
    mean_abs_delta: float

    @property
    def ok(self) -> bool:
        return (self.live_rows == self.shadow_rows
                and self.missing_keys == 0
                and self.max_abs_delta <= 35.0)   # tune per known re-weight


def check_parity(cur, window: BackfillWindow, old_version: int) -> ParityReport:
    cur.execute(
        """
        WITH live AS (
            SELECT fixture_id, capture_timestamp, registry_revision,
                   compliance_percentage AS old_pct
            FROM compliance_agg
            WHERE capture_timestamp >= %(start)s AND capture_timestamp < %(end)s
              AND registry_revision = ANY(%(revs)s) AND schema_version = %(old)s
        ),
        new AS (
            SELECT fixture_id, capture_timestamp, registry_revision,
                   compliance_percentage AS new_pct
            FROM compliance_agg_shadow
            WHERE schema_version = %(new)s
        )
        SELECT
            (SELECT count(*) FROM live)                                    AS live_rows,
            (SELECT count(*) FROM new)                                     AS shadow_rows,
            count(*) FILTER (WHERE n.new_pct IS NULL)                      AS missing_keys,
            coalesce(max(abs(l.old_pct - n.new_pct)), 0)                   AS max_delta,
            coalesce(avg(abs(l.old_pct - n.new_pct)), 0)                   AS mean_delta
        FROM live l
        LEFT JOIN new n USING (fixture_id, capture_timestamp, registry_revision);
        """,
        {"start": window.start, "end": window.end,
         "revs": list(window.registry_revisions),
         "old": old_version, "new": window.new_schema_version},
    )
    r = cur.fetchone()
    return ParityReport(r["live_rows"], r["shadow_rows"], r["missing_keys"],
                        float(r["max_delta"]), float(r["mean_delta"]))


def cutover(conn, window: BackfillWindow, old_version: int) -> None:
    """Atomically promote the shadow rows into the live series."""
    report = check_parity(conn.cursor(), window, old_version)
    if not report.ok:
        raise RuntimeError(f"parity failed, refusing cutover: {report}")
    with conn.cursor() as cur:
        cur.execute(
            """
            BEGIN;
            DELETE FROM compliance_agg
            WHERE capture_timestamp >= %(start)s AND capture_timestamp < %(end)s
              AND registry_revision = ANY(%(revs)s);
            INSERT INTO compliance_agg
            SELECT * FROM compliance_agg_shadow WHERE schema_version = %(new)s;
            COMMIT;
            """,
            {"start": window.start, "end": window.end,
             "revs": list(window.registry_revisions),
             "new": window.new_schema_version},
        )

The cutover runs inside one transaction so dashboards never observe a half-swapped window — readers see the old series or the new one, never a mix.

Verification & Testing

Confirm each guarantee deterministically rather than trusting a green run:

  1. Re-running is a no-op. Run run_backfill twice on the same window and assert the second call’s committed row count equals the first and that SELECT count(*) on the shadow table is unchanged — the upsert overwrote, it did not append.
  2. Row counts stay stable. After backfill, assert ParityReport.live_rows == shadow_rows and missing_keys == 0; a mismatch means the window or the batch iterator dropped keys.
  3. Spot parity holds. Pull 20 known fixtures across the window and assert each recomputed compliance_percentage matches a hand-run of the new score_fn on its raw payload, and that max_abs_delta sits within the band you expect from the re-weight.
  4. The boundary is not double-counted. Insert a capture at exactly end; assert it is excluded (the window is [start, end)) and that a second adjacent window starting at end scores it exactly once.
  5. A malformed payload is skipped, not fatal. Feed one raw record that raises in score_fn; assert the run completes, the bad key is logged, and every other fixture still lands.
  6. The snapshot restores. Simulate a bad cutover, restore compliance_agg_snapshot, and assert the live series byte-matches its pre-run state.

A healthy run reports missing_keys == 0, identical row counts, and a mean_abs_delta consistent with the magnitude of the scoring change — small for a field addition, larger but bounded for a genuine re-weight.

Troubleshooting

Symptom Likely root cause Remediation
Aggregate totals inflate on every re-run Writes are plain INSERT, or the conflict target omits part of the natural key Make the write ON CONFLICT (fixture_id, capture_timestamp, registry_revision, schema_version) DO UPDATE; assert the unique constraint exists
A midnight capture is scored twice Window upper bound is inclusive, so adjacent runs overlap at the boundary Use [start, end) — exclusive end — and keep all bounds UTC-aware; verify with test 4
Backfill stops halfway and leaves a partial series live Recompute wrote directly into the live table with no shadow/cutover gate Always land in compliance_agg_shadow and promote atomically only after ParityReport.ok; a crash then leaves the live series untouched
Charts shift by a whole hour after backfill Timezone skew — raw capture_timestamp stored naive or in local time, recompute assumes UTC Normalize all timestamps to UTC-aware at ingestion and in BackfillWindow; reject naive bounds in __post_init__
Batch host OOMs on a multi-year window Recompute pulls the whole range into memory, or paging uses OFFSET Stream with keyset-paged iter_raw_batches, bound batch_size (e.g. 5000), and commit per batch so memory stays flat
Parity flags a huge delta on every fixture New score_fn version mismatched, or comparing against the wrong old_version Confirm the pinned scoring version and the schema_version filters in check_parity; a uniform large delta is a bug, a legitimate re-weight moves scores unevenly
Back to top