glue-schema-evolution-framework

A contract driven ingestion framework that sorts every upstream schema change into 6 classified kinds before it lands, auto evolves the safe ones and quarantines the rest, at 110,154 rows/sec on DuckDB.

What this solves

• A 3am hard failure on a schema change that was actually safe. New nullable columns and int32 to int64 promotions are classified as safe and the contract auto evolves to the next version, including the physical ALTER COLUMN on the landing table. In the demo feed, 2 of the 10 batches evolve the contract from v1 to v3 without human intervention and land 1,968 of 2,000 rows each.
• A silent load that nulls a column and quietly breaks a report. A field that declares downstream_reports can never be dropped without a decision, even where standard BACKWARD compatibility would permit it. Batch 7 of the demo drops payment_method, and all 2,000 rows are held with the machine readable reason field_dropped:payment_method.
• A column that keeps its type but changes meaning. The drift detector runs a population stability index and a two sample Kolmogorov Smirnov test against the previous accepted batch. Batch 9 switches amount_usd from dollars to cents; the type check sees nothing, and the drift check reports PSI 12.43 with a KS p value that underflows to zero and a median that moved 92.50 to 9,447.50.

Executive summary

An upstream team changes a column and the change reaches a warehouse table nobody is watching. There are two ways this goes. The pipeline hard fails at 3am, somebody is paged, and the reports are stale until they finish reading a stack trace. Or, worse, the load succeeds: the column is nulled out or the units change, every job is green, and the finance dashboard shows a number that is wrong by a factor of 100 until somebody notices at the month end review. As a representative scenario, an analyst spending two days reconstructing a fortnight of misreported revenue, plus the decisions taken on those numbers in the meantime, is the kind of incident that costs far more than the pipeline that caused it. The second failure mode is the expensive one precisely because nothing alerts.

schema_guard puts a versioned contract between the source and the warehouse and makes the pipeline decide, per column, what a change means. Every difference between the registered contract and an incoming batch is classified into one of six kinds (additive safe, widening safe, narrowing breaking, type incompatible, field dropped, semantic drift) and resolved through a per field compatibility policy that keeps the Avro and Confluent meaning of BACKWARD, FORWARD, FULL and NONE. Safe changes evolve the contract and the physical table. Breaking changes hold the batch in a quarantine lane with the failing rule, the raw payload and a replay command. Rows that fail a value level rule are quarantined individually while the rest of the batch lands. The transforms are pure functions behind an AWS Glue shaped adapter, the landing zone is DuckDB modelling a Snowflake landing schema, and the equivalent Snowflake DDL is emitted with sqlglot and re-parsed in the test suite. Python, pandas, numpy and scipy do the work; there is no Spark and no cloud account anywhere in the runtime.

Measured on this repo, on a 2 vCPU container with 7.8 GB of RAM: a one million row batch ingests in 9,078 ms at the median of 5 repeats, which is 110,154 rows/sec, and contract checking accounts for 64.6 percent of that time. The same batch with the checks switched off runs at 222,811 rows/sec, so the honest headline is that safety roughly doubles ingest wall clock. What it buys on that batch is 16,000 defective rows held back instead of landed. The test suite is 201 tests at 93 percent statement coverage; every number in this README comes from benchmark/results/results.json or docs/terminal_capture.txt, both committed.

Architecture

flowchart TD
    subgraph sources["Source feeds (CSV, untyped)"]
        A1[billing_transactions]
        A2[crm_accounts]
        A3[web_events]
        A4[inventory_snapshots]
        A5[support_tickets]
    end

    A1 & A2 & A3 & A4 & A5 --> T

    subgraph glue["Glue shaped transform layer (pandas, pure functions)"]
        T["transform(dynamic_frame_like, ctx)<br/>normalise names, trim, drop reserved"]
    end

    R[("Contract registry<br/>contracts/&lt;dataset&gt;/vNNN.json<br/>immutable, per field policy")]

    T --> C
    R -.reads.-> C

    subgraph boundary1["Failure boundary 1: schema level"]
        C{"classify_batch<br/>6 change kinds x 4 policies"}
        D["drift: PSI + two sample KS<br/>vs last accepted batch"]
        C <--> D
    end

    C -->|"evolve: additive, widening"| E["write contract v+1<br/>ADD COLUMN / ALTER COLUMN"]
    C -->|"block: dropped, incompatible, drift"| QB["quarantine scope=batch<br/>nothing lands"]
    E --> V

    subgraph boundary2["Failure boundary 2: row level"]
        V{"validate_batch<br/>not null, cast, length, scale, key"}
    end

    V -->|"clean rows"| L
    V -->|"failing rows"| QR["quarantine scope=row<br/>rest of batch still lands"]

    L[("DuckDB landing zone<br/>LANDING.&lt;dataset&gt;<br/>merge on primary key")]

    QB --> RP{"replay:<br/>re-classify against<br/>current contract"}
    QR --> RP
    RP -->|"still breaking"| QB
    RP -->|"now clean"| L

    L --> DDL["sqlglot: Snowflake DDL<br/>artifacts/&lt;dataset&gt;.snowflake.sql"]
    L --> HTML["jinja2 batch inspection report"]

Loading

Tech stack

Technology	Role here	Why chosen for this problem
Python 3.11	The whole framework	The classifier is a decision table over types and distributions, which is ordinary application logic. Nothing here needs a distributed runtime.
pandas	Batch representation and every transform	The row rules are boolean masks over whole columns, so a million row batch is evaluated in seconds without a single Python loop. It is also the dataframe an AWS Glue Python shell job already has.
numpy	The PSI calculation and the rule masks	Quantile binning and histogram counts are one call each, and the masks that drive quarantine are numpy arrays so combining six rules costs nothing.
scipy	Two sample Kolmogorov Smirnov test	Drift needs a statistic with an interpretable threshold, not a hand rolled heuristic. KS is distribution free and returns a p value, which is what makes "below 0.01" a defensible cut off.
DuckDB	Local landing zone and quarantine store	The reviewer needs to run this with no account. DuckDB gives real SQL, real types that reject a bad cast, and a single file database, which is what makes the "guarded versus untyped" benchmark comparable.
sqlglot	Snowflake DDL emission	One statement is built once and rendered into both dialects, so the local table and the Snowflake artifact cannot drift apart. It is a parser, so it needs no Snowflake account to prove the output is valid.
jinja2	Self contained HTML batch report	The report is attached to a ticket or opened from object storage, so it must be one file with no CSS or JS fetched at render time.
rich	CLI tables	The person reading schema-guard ingest output at 3am needs the blocked batch to be visibly different from the landed ones.
PyYAML	Contract annotation specs	The annotations an analyst owns (which reports depend on a column, what unit it is in) belong in a file a human writes and reviews, separate from the generated contract JSON.
playwright	Screenshot rendering	Renders the real HTML report headlessly so the image in this README is the actual output, not a mock up.
pytest, pytest-cov, ruff	Tests and linting	201 tests at 93 percent, run on every push.

How this maps onto AWS Glue

This needs saying plainly, because the repo name mentions Glue and the code does not import awsglue.

What is real. Every transform in src/schema_guard/transforms.py has the signature a Glue PySpark job uses, transform(dynamic_frame_like, ctx) -> dynamic_frame_like. Frames travel in a DynamicFrameLike wrapper exposing the same fromDF / toDF / count / schema surface as awsglue.dynamicframe.DynamicFrame. TransformContext carries the job name, batch id, contract version and job arguments, which is what GlueContext plus getResolvedOptions provide. Every transform is a pure function of its inputs and returns a new frame, which is what makes them testable with a three row dataframe.

What is not real. The local runtime is pandas on a single process. There is no Spark, no Glue, no S3 and no AWS call anywhere in this repository, and none of the numbers in this README were measured on Glue.

What would change on Glue. Replace DynamicFrameLike with awsglue.dynamicframe.DynamicFrame and swap the pandas expression inside each transform for its Spark equivalent, roughly four lines per function: frame.rename(columns=...) becomes ApplyMapping or a withColumnRenamed chain, series.str.strip() becomes F.trim, frame.drop(columns=...) becomes .drop_fields. The classifier, the compatibility policy, the drift statistics, the quarantine store and the tests do not move: they operate on schemas and on aggregate statistics, not on the frame API. The row rules in validate.py would need porting to Spark column expressions, which is the largest single piece of work and is the honest reason this is described as a mapping rather than a port.

Quickstart

Prerequisites:

• Python 3.11 or newer
• git
• No cloud account, no credentials, no network access after install

git clone https://github.com/Sandeep0430/glue-schema-evolution-framework.git
cd glue-schema-evolution-framework

python3 -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"

# 1. Generate the five demo source feeds (10 batches each, deterministic).
schema-guard seed

# 2. Register contract v1 for each feed from its committed spec plus batch 1.
for ds in billing_transactions crm_accounts web_events inventory_snapshots support_tickets; do
  schema-guard register --dataset "$ds"
done

# 3. Ingest the whole billing story. Batches 3 and 5 evolve the contract,
#    batches 7 and 9 are blocked. Exit code 2 means a batch was held.
schema-guard ingest --dataset billing_transactions --all

# 4. Look at what is being held and why.
schema-guard quarantine summary --dataset billing_transactions

# 5. Replay before amending anything: the batch scoped holds stay put.
schema-guard replay --dataset billing_transactions

# 6. Amend the contract to accept the dropped column, then replay again.
schema-guard amend --dataset billing_transactions --field payment_method \
  --clear-downstream-reports --note "payment mix report retired, ticket DATA-812"
schema-guard replay --dataset billing_transactions

# 7. Emit the Snowflake DDL for the evolved table.
schema-guard ddl --dataset billing_transactions

# 8. Render the HTML batch inspection report for the drifted batch.
schema-guard report --dataset billing_transactions --batch 9

schema-guard status

Or with make: make setup && make demo && make test. Or with Docker: docker compose run --rm demo.

To reproduce the numbers in this README: make bench then make screenshots.

Screenshots

Batch inspection report

The jinja2 HTML report for billing_transactions-b009, rendered headless with playwright chromium at 1440x900. It shows the contract diff for the blocked batch (semantic_drift on amount_usd, DECIMAL(12,2) unchanged, action block), the drift statistics behind that verdict (PSI 12.431, KS statistic 0.999, median moved 92.50 to 9,447.50, a ratio of 102.1x), the full quarantine queue broken down by rule and scope, and the batch history showing where the contract evolved from v1 to v3.

Ingest throughput and the cost of contract checking

Throughput in rows per second against batch size for the guarded and untyped paths, with the contract checking overhead plotted underneath as its own series in percent. Both the attributed share (time inside classify_batch plus validate_batch) and the wall clock share (guarded run measured against the baseline) are shown, because they answer different questions. Generated from benchmark/results/results.json by benchmark/plot_results.py.

The CLI catching the breaking change, and the test suite

Real stdout, captured to docs/terminal_capture.txt and rendered to PNG. Batches 7 and 9 are blocked with 2,000 rows each routed to quarantine, the classified change and its justification are printed for both, inspect re-runs the batch 9 verdict without landing anything and shows PSI 12.431 with KS p=0.00e+00 and a median that moved 92.5 to 9,448, the quarantine summary shows the six distinct rules holding rows, replay declines to release either batch scoped hold, and pytest reports 201 passed at 93 percent coverage.

Performance under load

Method: benchmark/run_benchmark.py ingests one generated billing_transactions batch at 10k, 100k and 1M rows, 5 repeats per scale after one discarded warmup pass per path, with the guarded and baseline paths alternating order so neither pays for the other's cache warming. Each run gets a fresh contract registry and a fresh DuckDB file. The baseline is the same batch landed into an all text table with no declared key, no classifier and no row rules, which is what schema on read actually does. Container: 2 logical CPUs, 7.8 GB RAM, Linux 6.18.5, python 3.11.15, duckdb 1.5.5, pandas 3.0.2, numpy 2.4.4.

Rows	Guarded p50	p95	p99	Guarded rows/sec	Baseline rows/sec	Checks, attributed	Checks, wall clock	Rows quarantined
10,000	167 ms	169 ms	169 ms	59,795	131,901	49.1%	120.6%	160
100,000	936 ms	967 ms	973 ms	106,856	243,558	60.1%	127.9%	1,600
1,000,000	9,078 ms	9,152 ms	9,156 ms	110,154	222,811	64.6%	102.3%	16,000

Where it degrades: the attributed cost of checking climbs from 49.1 to 64.6 percent as the batch grows, because the fixed costs it competes with (opening DuckDB, creating the table, loading the contract) stay constant while the per row work does not. Classification is the larger half, not the row rules: at one million rows it is 3,675 ms against 2,189 ms, because type inference coerces all eight columns of untyped text before anything is compared. Both halves scale linearly, so a batch beyond a few million rows on this container is bounded by pandas doing several full column passes in a single process. The fix is chunking, and caching inference per column would take the larger bite; both are out of scope below.

Architecture Decision Records

• ADR 0001: A file based contract registry with an explicit per field compatibility policy. Why contracts are immutable JSON in the repo rather than a Confluent Schema Registry dependency or schema on read, and why a field that names its downstream reports is treated more strictly than Avro would.
• ADR 0002: Quarantine and continue, not fail fast. The boring choice, with the honest consequence written down: a quarantine queue nobody watches is worse than a failure, because it also feels fine. Covers the three mitigations built in and the one deliberately left out.

Intentionally out of scope

• Streaming ingestion. Everything here is batch, and the compatibility semantics are applied to a batch rather than a message. Add a streaming path when a source starts emitting to Kafka and the contract has to be enforced per message.
• Chunked ingestion for very large batches. Batches are held in memory. Add chunking when a single batch exceeds roughly a quarter of container memory, which on the benchmark container is somewhere past 5 million rows.
• Alert routing. Quarantine depth is exposed through the CLI, the ingest log table and WARNING level structured logs, and stops there. Add an integration when the team agrees where alerts should land; see ADR 0002 for why a half built one would be worse than none.
• Cross process locking on the registry. Two concurrent ingests of the same dataset can race on the next contract version. Add locking when ingestion stops being serialised per dataset by the scheduler.
• Automatic contract amendment. amend is a deliberate human action for a reason. Add a proposal mechanism (open a pull request with the suggested contract diff) before adding anything that applies one automatically.
• Column level lineage. downstream_reports is a hand maintained annotation, not a parsed dependency graph. Wire it to a real lineage source when one exists, because the annotation going stale is the most likely way this framework starts lying.

Security and compliance

• No credentials exist in this project. There is no cloud client, no connection string and no auth code. Settings.describe() returns only paths and thresholds and is the only thing that prints configuration.
• Configuration is environment only. Every tunable is read through schema_guard.config from a SCHEMA_GUARD_ prefixed variable, documented with a safe default in .env.example. Nothing is hardcoded and .env is gitignored.
• What is never logged. Structured log lines carry counts, rule ids, column names, batch ids and the run id. Row values never appear in a log line. Values are written only to the quarantine table, inside the same database as the data itself, so quarantine inherits the warehouse's access controls rather than leaking into a log aggregator with different ones.
• Quarantine is data, and holds source records verbatim. If a source carries personal data, the quarantine table does too. It must be covered by the same retention and access policy as the landing tables, and the raw payload column is the reason.
• Least privilege on the target. The emitted Snowflake DDL only ever creates a schema, creates tables and adds or widens columns. Nothing generated by this framework drops a column, drops a table or deletes outside the primary key merge, so the deploying role needs CREATE and ALTER on one schema and nothing more.
• No network at runtime. sqlglot transpiles locally, playwright renders a local file, and the whole test suite and demo run offline.

Failure modes

Failure	Detection	Behaviour	Recovery
Upstream drops a column a report depends on	classify_batch reports field_dropped and the field declares downstream_reports	Whole batch quarantined with scope=batch, contract unchanged, CLI exits 2, ERROR log line	schema-guard amend --field X --clear-downstream-reports after confirming the report is retired, then schema-guard replay
Amount column switches units (dollars to cents)	PSI at or above 0.25 and KS p below 0.01 against the last accepted batch	Whole batch quarantined as semantic_drift, nothing lands, contract unchanged	Confirm the unit change upstream, amend the contract's unit annotation and scale, then replay. Until then the wrong numbers are held, not landed
A few rows have nulls, bad casts or duplicate keys	Row rules produce a per row rule id such as not_null:currency	Those rows quarantined with scope=row, the other 98.4 percent of the batch lands	Fix at source and re-ingest, or schema-guard replay after amending the contract if the rule was too strict
Contract widens a column but the physical table does not	DuckDB raises an out of range conversion on insert	The batch fails loudly rather than truncating	Fixed in cc06db6: evolve_table now issues ALTER COLUMN for any contract widening, gated on is_widening so it can never narrow
The same bad record arrives every day	Quarantine id is sha256(dataset, rule, payload) and excludes the batch id	The second and later arrivals are recognised as already present and do not grow the queue	schema-guard quarantine list shows one row with the original batch id, which is what identifies the source of the repeat
Replay would bypass the schema gate	Batch scoped holds are re-classified against the current contract before release	A hold whose change is still breaking stays held, and the reason is printed per batch	Amend the contract for that specific column. An amendment to a different column releases nothing
Two ingests race on the same contract version	ContractRegistry.save refuses to overwrite an existing version file	The second writer raises a ContractError and its batch does not land	Serialise ingestion per dataset in the scheduler, then re-run the losing batch. See ADR 0001
Quarantine grows and nobody looks	schema-guard status reports open depth per dataset; every write emits a WARNING log	The pipeline stays green, which is the risk	Alert on quarantine depth. This is the acknowledged cost of ADR 0002 and the first metric to watch after deploying

Hardest problem solved

The classifier was working, the policy table was working, and the demo died on batch 5 with _duckdb.ConversionException: Conversion Error: Type INT64 with value 2216450393 can't be cast because the value is out of range for the destination type INT32 when casting from source column account_id. The confusing part was that the framework had already made the right call. classify_batch returned widening_safe for account_id, the BACKWARD policy resolved it to evolve, and contract v3 was sitting on disk saying BIGINT. The registry and the error disagreed about what the column was.

The root cause was that contract evolution and physical table evolution were two separate things that nobody had connected. LandingZone.evolve_table compared the contract's field names against the table's column names and issued ADD COLUMN for anything missing. It never compared types, so ddl.build_alter_type had been written and had zero callers. The registry said BIGINT, DuckDB still said INTEGER, and the framework's central claim, that widening is safe because it is handled, was false at exactly the layer where it mattered. It only surfaced at batch 5 because that is the first batch where the source key space actually crosses 2^31; every earlier batch fit in an INTEGER and the mismatch was invisible.

The fix, in cc06db6, makes evolve_table read the physical types back out of information_schema, map them into the same logical lattice the classifier uses through a new duckdb_type_to_spec, and issue ALTER COLUMN ... SET DATA TYPE for any field whose contract type is a strict widening of what the table holds. The promotion is gated on is_widening, so this path can never narrow a column and truncate landed history; a narrowing has to go through a contract amendment that a person approved. The lesson generalised beyond the bug: a guarantee that stops at the metadata layer is not a guarantee, and the test that now covers it (test_evolve_table_promotes_a_widened_column) asserts the physical information_schema type rather than the contract, because asserting the contract is what let this through in the first place.

Future work

• Quarantine depth as a first class metric. The first thing to watch after deploying is open quarantine rows per dataset per day, with an alert on growth rather than on level. Level is noisy; sustained growth means an upstream change nobody has triaged.
• Chunked ingestion with a streaming validator. Batches currently load whole. Evaluating the row rules over chunks would flatten memory and let the same code handle a batch an order of magnitude larger, at the cost of making the primary key duplicate rule stateful.
• Drift on derived measures, not just raw columns. A currency switch is caught because the column moved. A change in the mix of a categorical dimension that shifts a weighted average is not, because no single column's distribution moves enough. Testing a small set of declared business measures alongside the raw columns would close that gap.
• A dbt exposure importer for downstream_reports. The annotation is hand maintained today, and a stale annotation is the most likely way this framework starts lying about what is safe to drop. Reading dbt exposures would make it derived rather than declared.
• Contract proposals as pull requests. When the classifier resolves a change to block, it already knows what contract would accept it. Emitting that as a branch and a pull request would keep the human decision while removing the manual editing.