Detecting silent schema drift in municipal feeds

The most dangerous failure in a municipal ingestion pipeline is the one that does not raise an exception. A county quietly renames ZONECODE to zone_class between two releases, changes a parcel AREA from an integer to a float, or swaps a layer’s geometry from Polygon to MultiPolygon — and ships it with no changelog. Your loader, written to be tolerant, maps whatever columns it finds, casts whatever types arrive, and commits a batch that validates clean while its meaning is corrupted: acreage now off by orders of magnitude, districts silently NULLed, dissolves producing wrong boundaries. Nothing throws, so nothing alerts, and the bad data flows downstream into every report. This guide, part of the schema validation and data quality checks topic within the municipal zoning data architecture and compliance frameworks area, shows how to detect that drift the only way that reliably works: by capturing a schema fingerprint per source and diffing every incoming release against it before the load runs.

Diagnosis: why tolerant loaders hide drift jump to heading

A pipeline built to survive messy municipal data is, by construction, built to not notice when that data changes shape. Three drift classes account for nearly every silent corruption incident, and each defeats a different naive assumption.

1. Column rename. The feed drops ZONECODE and introduces zone_class carrying the same values. A loader that maps columns by fuzzy name match or position happily binds the new column — or, worse, finds no ZONECODE, fills the target with NULL, and commits. No row count changes; the district field just goes empty.
2. Type change. parcel_area arrives as a float where it was an integer, or a date column becomes an ISO string. Permissive casting (float(x), pd.to_numeric(errors="coerce")) swallows it, and a coerce-to-NaN on an unparseable value silently deletes data one cell at a time.
3. Geometry-type swap. A layer previously Polygon ships as MultiPolygon (or gains a Z coordinate, or flips ring winding). Shapely reads both, so the load succeeds, but any code assuming single-part geometry — a centroid, an area sum, a dissolve — now computes against a different structure and produces plausible-but-wrong numbers.

The unifying property is that none of these produces an error. You cannot catch them with a try/except because nothing is thrown; you cannot catch them with a row-count check because the row count is unchanged. You catch them by comparing the structure of this release to a recorded contract of what the last known-good release looked like. Reproduce the blind spot before fixing it: load last quarter’s extract and this quarter’s, print dict(df.dtypes) and df.geometry.geom_type.unique() for each, and watch a clean load hide a changed schema.

Step-by-step implementation jump to heading

Each step isolates one concern; the final step wires them into the load gate.

Step 1 — Capture a schema contract per source jump to heading

A schema contract is a compact, comparable record of a source’s structure: column names, dtypes, geometry types, and CRS. Compute a stable hash over the normalized structure so a single equality check tells you whether anything moved, then keep the full detail for diffing. Store one contract per source, not one global schema — each county drifts on its own clock.

import hashlib
import json
import geopandas as gpd

def fingerprint_schema(gdf: gpd.GeoDataFrame) -> dict:
    """Build a comparable schema contract from a loaded feed."""
    # Sort so field order changes do not falsely register as drift.
    columns = {col: str(dtype) for col, dtype in sorted(gdf.dtypes.items())}
    geom_types = sorted(gdf.geometry.geom_type.dropna().unique().tolist())
    contract = {
        "columns": columns,                         # name -> dtype string
        "geometry_types": geom_types,               # e.g. ["MultiPolygon"]
        "crs": gdf.crs.to_epsg() if gdf.crs else None,
    }
    # Deterministic hash: same structure -> same digest, order-independent.
    blob = json.dumps(contract, sort_keys=True).encode()
    contract["schema_hash"] = hashlib.sha256(blob).hexdigest()
    return contract

Step 2 — Diff each release against the last known-good contract jump to heading

On every run, fingerprint the incoming release and compare it to the stored contract for that source. The fast path is the hash: if the digests match, the structure is identical and you can skip detailed comparison. When they differ, compute the exact deltas so the next step can classify them.

def diff_schema(old: dict, new: dict) -> dict:
    """Return structural deltas between two schema contracts."""
    if old["schema_hash"] == new["schema_hash"]:
        return {"changed": False}

    old_cols, new_cols = old["columns"], new["columns"]
    old_keys, new_keys = set(old_cols), set(new_cols)

    added = sorted(new_keys - old_keys)
    removed = sorted(old_keys - new_keys)
    retyped = sorted(
        c for c in (old_keys & new_keys) if old_cols[c] != new_cols[c]
    )
    geom_changed = old["geometry_types"] != new["geometry_types"]
    crs_changed = old["crs"] != new["crs"]

    return {
        "changed": True,
        "added": added, "removed": removed, "retyped": retyped,
        "geometry_changed": geom_changed, "crs_changed": crs_changed,
    }

Step 3 — Classify additive versus breaking drift jump to heading

Not all drift is equal. A brand-new nullable column is additive — it does not invalidate existing mappings and can pass automatically while you record the new contract. A removed or renamed column, a type change, a geometry-type swap, or a CRS change is breaking — it corrupts meaning and must stop the load. Renames surface here as a simultaneous add plus remove; treat any removal as breaking so a renamed ZONECODE cannot slip through as a harmless addition.

def classify_drift(delta: dict) -> str:
    """Return 'none', 'additive', or 'breaking' for a schema diff."""
    if not delta["changed"]:
        return "none"
    breaking = (
        delta["removed"]           # dropped or renamed column
        or delta["retyped"]        # int -> float, str -> date, etc.
        or delta["geometry_changed"]
        or delta["crs_changed"]
    )
    if breaking:
        return "breaking"
    # Only new columns appeared and nothing else moved.
    return "additive" if delta["added"] else "none"

Step 4 — Gate the load and alert jump to heading

Wire classification into the load path so a breaking change refuses to commit. An additive change passes but updates the stored contract so the new column becomes the baseline; a breaking change raises and routes to the alert channel with the exact deltas, which is far more actionable than a generic validation failure. Normalizing the newly added or renamed fields into your canonical model is the job of the attribute normalization rules layer once a human has approved the change.

import logging

logger = logging.getLogger("schema_drift")

class SchemaDriftError(Exception):
    """Raised when a breaking structural change is detected."""

def gate_load(source_id: str, gdf, contract_store, alerter) -> None:
    new_fp = fingerprint_schema(gdf)
    old_fp = contract_store.get(source_id)

    if old_fp is None:                      # first sighting: adopt as baseline
        contract_store.put(source_id, new_fp)
        logger.info(f"{source_id}: captured initial schema contract")
        return

    delta = diff_schema(old_fp, new_fp)
    verdict = classify_drift(delta)

    if verdict == "breaking":
        alerter.send(f"BREAKING schema drift on {source_id}: {delta}")
        raise SchemaDriftError(f"{source_id} breaking drift: {delta}")

    if verdict == "additive":
        logger.warning(f"{source_id}: additive drift {delta['added']} — baseline updated")
        contract_store.put(source_id, new_fp)   # accept and move baseline forward
    # verdict == "none": structure unchanged, proceed silently.

Step 5 — Quarantine drifted batches jump to heading

A breaking release must not vanish and must not commit. Write the whole batch to a quarantine partition alongside its fingerprint and the diff, so a human can review the change, decide whether the rename is legitimate, and re-drive it through a contract update. This mirrors the dead-letter discipline used across the ingestion layer and keeps the run reproducible.

from datetime import datetime, timezone
from pathlib import Path

def quarantine_batch(source_id: str, gdf, delta: dict, root="quarantine") -> Path:
    """Persist a drifted batch and its diff for human review."""
    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    out_dir = Path(root) / source_id / stamp
    out_dir.mkdir(parents=True, exist_ok=True)
    # Keep geometry + CRS intact so the batch can be replayed after approval.
    gdf.to_parquet(out_dir / "batch.parquet", index=False)
    (out_dir / "drift.json").write_text(json.dumps(delta, indent=2, sort_keys=True))
    logger.error(f"{source_id}: quarantined drifted batch to {out_dir}")
    return out_dir

Verification & testing jump to heading

Confirm the gate catches real drift and does not fire on noise.

• Rename is caught as breaking. Fixture two extracts identical except ZONECODE renamed to zone_class; assert classify_drift returns breaking and the load raises. A rename passing as additive means removals are not being treated as breaking.
• Additive column passes and moves the baseline. Add one nullable column to an otherwise identical extract; assert the load proceeds and the stored contract now includes it, so the same column does not re-alert next run.
• Hash is order-independent. Reorder the columns of an unchanged extract and assert schema_hash is identical. If it changes, the fingerprint is not sorting fields and every reorder will false-positive.
• Geometry swap trips the gate. Feed a MultiPolygon release where the contract recorded Polygon; assert geometry_changed is true and the verdict is breaking, since single-part assumptions downstream would otherwise silently misbehave.
• Idempotent no-op. Re-running the same release twice must classify as none both times and never re-quarantine, proving the baseline is being read, not recomputed from the incoming batch.

Failure recovery jump to heading

• Approve a legitimate rename. When review confirms the county genuinely renamed a field, update the source’s contract to the new fingerprint and add the alias to the normalization map so historical ZONECODE and new zone_class resolve to the same canonical field, then replay the quarantined batch through the gate.
• Replay from quarantine. Quarantined batches carry their geometry, CRS, and diff, so recovery is a re-drive against the updated contract — never a hand-edit of the target table. Keep replays idempotent by keying on the source hash so a re-run cannot double-write.
• Backfill the corruption window. If drift slipped through before the gate existed, identify the first bad release by walking stored fingerprints, then reload every affected batch from quarantine or source. Downstream change detection and geometry diffing will surface exactly which parcels flipped during the window so you can scope the backfill precisely.
• Tune the additive threshold. If a chatty source adds columns constantly and floods the alert channel, keep additive changes silent-but-logged and reserve alerts for breaking verdicts, rather than loosening the breaking classification and letting real corruption through.

Frequently asked questions jump to heading

Related jump to heading

• Parent topic: Schema Validation & Data Quality Checks
• Section overview: Municipal Zoning Data Architecture & Compliance Frameworks
• Attribute Normalization Rules — mapping renamed and new fields into the canonical model after approval
• Change Detection & Geometry Diffing — scoping which parcels a drift window actually corrupted