Python: Data Warehouse – Catching Schema Drift Before an ETL Run

The ETL failures that cost the most time are never the ones that crash loudly. They are the ones where a source table quietly loses a column, or a varchar(50) becomes varchar(255), and the load "succeeds" while writing garbage, truncated values, or nulls downstream. I run this script as a pre-flight check before every scheduled ETL job: snapshot the expected schema once, commit it, then diff the live schema against it on every run and fail fast if something moved.

This version compares more than column names and types. It also compares length, numeric and timestamp precision, interval fields, nullability, the schema-qualified underlying type for arrays and user-defined types, and the domain a column uses. It sorts findings into breaking changes (a column removed, a type changed, a length or precision reduced, a column becoming nullable) and additive ones (a new column, a length increased). A pipeline that selects an explicit column list, like the incremental Postgres-to-BigQuery load, usually survives additive changes, so --allow-additive lets those pass with a warning.

Requirements

Python 3.10 or later.
psycopg2-binary 2.9.1 or later (pip install psycopg2-binary), the first release with Python 3.10 wheels.
A role that can see the tables. information_schema.columns only shows columns the current user has access to, as owner or through some privilege. A table the ETL role can't read looks exactly like a table that was dropped, which is what you want the check to catch anyway.
A baseline snapshot generated once with --snapshot, committed to version control next to the pipeline that depends on it.

Usage

Take the baseline the first time and commit the file. Table names can be schema-qualified; unqualified names default to public.

bash

python check_schema_drift.py --snapshot \
    --dsn "postgresql://etl_reader:<password>@<host>:5432/app" \
    --tables public.orders,public.customers,billing.line_items \
    --output schema_baseline.json

The baseline records every attribute that is compared:

json

{
  "public.orders": {
    "discount_code": {
      "character_maximum_length": 32,
      "data_type": "character varying",
      "datetime_precision": null,
      "domain_name": null,
      "domain_schema": null,
      "interval_type": null,
      "is_nullable": "YES",
      "numeric_precision": null,
      "numeric_scale": null,
      "udt_name": "varchar",
      "udt_schema": "pg_catalog"
    },
    "total_cents": {
      "character_maximum_length": null,
      "data_type": "integer",
      "datetime_precision": null,
      "domain_name": null,
      "domain_schema": null,
      "interval_type": null,
      "is_nullable": "NO",
      "numeric_precision": 32,
      "numeric_scale": 0,
      "udt_name": "int4",
      "udt_schema": "pg_catalog"
    }
  }
}

Run the check as the first step of the job and chain the load after it, so a non-zero exit stops the pipeline before any extraction:

bash

python check_schema_drift.py \
    --dsn "$SOURCE_DSN" \
    --tables public.orders,public.customers,billing.line_items \
    --baseline schema_baseline.json \
    --allow-additive \
  && python postgres_to_bigquery_incremental.py --pg-dsn "$SOURCE_DSN" --source-table public.orders ...

Sample output when drift is found:

text

BREAKING billing.line_items: sku: character_maximum_length 64 -> 32
ADDITIVE public.customers: column added: loyalty_tier (character varying)
BREAKING public.orders: column removed: discount_code
BREAKING public.orders: total_cents: type int4 -> numeric
3 breaking, 1 additive change(s). Exit 1.

With only additive changes and --allow-additive, it prints the same ADDITIVE lines and exits 0.

Script

python

#!/usr/bin/env python3
"""
check_schema_drift.py

Compares live PostgreSQL column definitions against a saved baseline and
reports added, removed, retyped (including a change of domain or of the
schema a user-defined type lives in), resized, precision-changed, or
nullability-changed columns before an ETL job runs.

Usage:
    Baseline: python check_schema_drift.py --snapshot --dsn <dsn> --tables s.t1,t2 --output baseline.json
    Check:    python check_schema_drift.py --dsn <dsn> --tables s.t1,t2 --baseline baseline.json [--allow-additive]

Reads:  information_schema.columns for each table through the given DSN.
Writes: a JSON snapshot in --snapshot mode; nothing in check mode.
Exit:   0 no drift (or only additive drift with --allow-additive),
        1 drift found, 2 usage or connection error.
"""

import argparse
import json
import sys

import psycopg2

# The first five together identify the column's type; the rest are compared one by one.
TYPE_ATTRIBUTES = ("data_type", "udt_schema", "udt_name", "domain_schema", "domain_name")
DETAIL_ATTRIBUTES = (
    "character_maximum_length",
    "numeric_precision",
    "numeric_scale",
    "datetime_precision",
    "interval_type",
    "is_nullable",
)
ATTRIBUTES = TYPE_ATTRIBUTES + DETAIL_ATTRIBUTES

COLUMN_QUERY = """
    SELECT column_name, data_type, udt_schema, udt_name, domain_schema, domain_name,
           character_maximum_length, numeric_precision, numeric_scale,
           datetime_precision, interval_type, is_nullable
    FROM information_schema.columns
    WHERE table_schema = %s AND table_name = %s
    ORDER BY ordinal_position
"""


def qualify(name: str) -> tuple[str, str]:
    """Split 'schema.table' (default schema public) into its parts."""
    schema, _, table = name.rpartition(".")
    return (schema or "public", table)


def fetch_schema(dsn: str, tables: list[str]) -> dict[str, dict[str, dict]]:
    """Return {'schema.table': {column: {attribute: value}}}. Missing tables map to {}."""
    result: dict[str, dict[str, dict]] = {}
    conn = psycopg2.connect(dsn)
    try:
        with conn, conn.cursor() as cursor:
            for name in tables:
                schema, table = qualify(name)
                cursor.execute(COLUMN_QUERY, (schema, table))
                columns = {}
                for row in cursor.fetchall():
                    column_name, *values = row
                    columns[column_name] = dict(zip(ATTRIBUTES, values))
                result[f"{schema}.{table}"] = columns
    finally:
        conn.close()
    return result


def type_label(column: dict) -> str:
    """Domain name if the column uses one, else the underlying type; schema shown unless pg_catalog."""
    schema = column.get("domain_schema") or column.get("udt_schema")
    name = column.get("domain_name") or column.get("udt_name")
    return name if schema in (None, "pg_catalog") else f"{schema}.{name}"


def classify(attribute: str, old, new) -> str:
    """Return 'additive' for widening changes, 'breaking' for everything else."""
    widening = ("character_maximum_length", "numeric_precision", "datetime_precision")
    if attribute in widening and old is not None and new is not None and new > old:
        return "additive"
    if attribute in widening and old is not None and new is None:
        return "additive"  # length limit removed, e.g. varchar(50) -> varchar
    if attribute == "is_nullable" and old == "YES" and new == "NO":
        return "additive"  # stricter source never produces a value the load can't take
    return "breaking"


def diff_schema(baseline: dict, current: dict) -> list[tuple[str, str, str]]:
    """Return [(severity, table, message)] for every difference."""
    findings = []
    for table in sorted(set(baseline) | set(current)):
        old_cols = baseline.get(table, {})
        new_cols = current.get(table, {})

        if old_cols and not new_cols:
            findings.append(("breaking", table, "table missing or not visible to this role"))
            continue
        if new_cols and not old_cols:
            findings.append(("additive", table, "table not in baseline"))
            continue

        for column in sorted(set(old_cols) - set(new_cols)):
            findings.append(("breaking", table, f"column removed: {column}"))
        for column in sorted(set(new_cols) - set(old_cols)):
            findings.append(("additive", table, f"column added: {column} ({new_cols[column]['data_type']})"))
        for column in sorted(set(old_cols) & set(new_cols)):
            old_type = tuple(old_cols[column].get(a) for a in TYPE_ATTRIBUTES)
            new_type = tuple(new_cols[column].get(a) for a in TYPE_ATTRIBUTES)
            if old_type != new_type:
                # One finding per retyped column; precision/length differences follow from it.
                old_label, new_label = type_label(old_cols[column]), type_label(new_cols[column])
                findings.append(("breaking", table, f"{column}: type {old_label} -> {new_label}"))
                continue
            for attribute in DETAIL_ATTRIBUTES:
                old = old_cols[column].get(attribute)
                new = new_cols[column].get(attribute)
                if old != new:
                    severity = classify(attribute, old, new)
                    findings.append((severity, table, f"{column}: {attribute} {old} -> {new}"))
    return findings


def main() -> int:
    parser = argparse.ArgumentParser(description="Detect PostgreSQL schema drift before an ETL run.")
    parser.add_argument("--dsn", required=True, help="libpq connection string or URI")
    parser.add_argument("--tables", required=True, help="Comma-separated tables, optionally schema-qualified")
    parser.add_argument("--snapshot", action="store_true", help="Write a baseline instead of checking")
    parser.add_argument("--output", default="schema_baseline.json", help="Baseline path to write (--snapshot)")
    parser.add_argument("--baseline", help="Baseline path to compare against")
    parser.add_argument("--allow-additive", action="store_true", help="Exit 0 when all drift is additive")
    args = parser.parse_args()

    tables = [t.strip() for t in args.tables.split(",") if t.strip()]
    try:
        current = fetch_schema(args.dsn, tables)
    except psycopg2.Error as error:
        print(f"error: could not read schema: {error}", file=sys.stderr)
        return 2

    if args.snapshot:
        empty = [t for t, cols in current.items() if not cols]
        if empty:
            print(f"error: no visible columns for {', '.join(empty)}", file=sys.stderr)
            return 2
        with open(args.output, "w", encoding="utf-8") as handle:
            json.dump(current, handle, indent=2, sort_keys=True)
            handle.write("\n")
        print(f"Baseline for {len(current)} table(s) written to {args.output}")
        return 0

    if not args.baseline:
        print("error: --baseline is required unless --snapshot is given", file=sys.stderr)
        return 2

    with open(args.baseline, encoding="utf-8") as handle:
        baseline = json.load(handle)
    # Only compare the tables requested on this run.
    baseline = {t: cols for t, cols in baseline.items() if t in current}

    findings = diff_schema(baseline, current)
    if not findings:
        print("No schema drift detected.")
        return 0

    for severity, table, message in findings:
        print(f"{severity.upper()} {table}: {message}")

    breaking = sum(1 for f in findings if f[0] == "breaking")
    additive = len(findings) - breaking
    status = 0 if breaking == 0 and args.allow_additive else 1
    print(f"{breaking} breaking, {additive} additive change(s). Exit {status}.")
    return status


if __name__ == "__main__":
    sys.exit(main())

Notes

Why table_schema matters. Filtering information_schema.columns only on table_name is a common mistake. With an orders table in both public and an archive schema, it merges the two column lists, which can hide drift or invent it. Always filter on both.
Why udt_name as well as data_type. For arrays, data_type is just ARRAY, and for enums and other user-defined types it is USER-DEFINED. The real type is in udt_name, and udt_schema says which schema it lives in, so two enums called status in different schemas don't compare equal. Without them, changing a column from text[] to integer[], or from one enum to another, doesn't show up. For a domain, data_type and udt_name report the underlying type, so the script also compares domain_schema and domain_name: moving a column between two domains over the same base type is a type change too.
Precision beyond length. datetime_precision catches timestamp(6) becoming timestamp(0) (same udt_name, fewer fractional digits), and interval_type catches an interval column restricted to, say, DAY TO SECOND.
Length is where silent truncation lives. character_maximum_length is null for unbounded varchar and text, and set for varchar(n). A source column widened from 64 to 255 is harmless to read, but a destination still declared at 64 will reject the longer values. That's why the script flags it: the next step is to check the destination, not to suppress the warning.
Nullability direction. is_nullable going from NO to YES is breaking because the source can now send nulls into a destination column declared NOT NULL, or into metrics that assumed a value. The reverse is harmless for the load. Keep in mind that Postgres reports YES whenever a column is possibly nullable.
What it doesn't check. Defaults, constraints, indexes, and whether a column's meaning changed. A status column that gains a new value passes this check; data tests (dbt's accepted_values, or a plain SELECT DISTINCT assertion) cover that.
Treat the baseline like a migration. Update it deliberately, in the same commit as the pipeline change that expects the new shape, not as a reflex when the check goes red. Re-running --snapshot to make a failing check pass throws away exactly the information the check exists to surface.

Python: Data Warehouse – Catching Schema Drift Before an ETL Run

$ grep -n '^#' post.md

Requirements

Usage

Script

Notes

References