--- title: 'Schema Contracts' source: 'https://academia.sh/en/courses/introduction-to-data-engineering/schema-contracts' course: 'Introduction to Data Engineering' language: en updated: '2026-10-21T05:03:01+00:00' license: 'CC BY-SA 4.0' objectives: - 'Writes a schema contract as a running predicate that declares field names, types and the behavior for extra fields.' - 'Measures separately, for a field rename, the number of faulty rows on the producer side and the number of rows dropped on the consumer side.' - 'Compares a permissive and a strict contract on the same harmless schema change and counts the good rows the strict contract falsely blocks.' --- # Schema Contracts Counting whom the schema contract between producer and consumer breaks and what the contract itself costs: when a field is renamed, the producer side has 0 faulty rows while on the consumer side 200 rows are dropped at screening, 200 more at consumption, and the summary goes from 55 cells to 45; once the contract is written the same corruption is caught at stage 0 but the output is not repaired; a strict contract that pins the field set exactly blocks 1200 good rows with no corruption present and drops the summary to 25 cells; batch-level enforcement blocks 2200 good rows and yields no output at all, while an adapter stage returns the serving to the baseline and grows the pipeline to six stages. The fastest-caught corruption of the previous lesson was `schema_changed`: when the field name changed, the `schema` check fired at stage 0. The field list that check looked at was buried inside the code, and who wrote the list was not asked. The list is the consumer's assumption; the producer knows nothing about it. This lesson turns that list into a **data contract**. Three things are measured: on whose side how many rows are dropped when a field name changes, at which stage the corruption is caught once the contract is written, and what the contract itself costs. The third number is the most important, because it is an example of the second claim of the course: **adding a check also breaks things.** - **QP8.** The source, the pipeline and the contract are **fictional**; no real database, file format or orchestrator is called. The seed is **20260218**, and the measurement does not depend on the run. - **QP9.** The relational schema, normalization, constraint and type procedures were measured in the Databases curriculum, in the Data Modeling and Relational Theory course, and **are not repeated.** The question here is not how a schema is designed but **who breaks when it changes**. - **QP10.** **A schema change is not an attack.** A producer renames a field to make it clearer or adds a new field; both are ordinary maintenance work. - **QP11.** The contract runs **at stage 0**, before the source enters the pipeline, and is enforced at row level: a row that does not conform is blocked, a row that conforms passes. - **QP12.** The baseline is the **pipeline without a contract**: blocked rows **0**, corruption caught **0**. - **QP13.** Resolution: the step of the row share is **0.0004**, of a summary cell **0.0182**. **No share is printed on the serving output**, the value itself is written. ## A Contract Is Not a List of Fields A list of fields only says "let these names exist". A contract has to declare three things at once: which fields **exist**, what the **type** of each is, and what will happen when an undeclared field arrives. The third is mostly not written, and the most expensive number of this lesson comes from there. The difference between the two implementations is a single line. A **permissive** contract requires the declared fields to exist and their types to match; it leaves an extra field free. A **strict** contract requires the field set to be **exactly equal** to the declared one; an extra field is also a violation. Both read the same contract text and behave the same on a corrupted source; where they part ways is a **harmless** change. ```python # FICTIONAL. Meter readings, the five-stage pipeline and the contract are defined in the lesson; # no real store, file format or orchestrator is called. SEED, M32 = 20260218, 0xFFFFFFFF def rng(t): x = ((t ^ (t >> 16)) * 2246822507) & M32 x = ((x ^ (x >> 13)) * 3266489909) & M32 s = [(x ^ (x >> 16)) & M32] def next_value(): s[0] = (s[0] * 1664525 + 1013904223) & M32 return s[0] / 4294967296 return next_value REGION = ["east", "west", "north", "south", "central"] TYPE = ["residential", "commercial"] PERIODS = [f"2026-{a:02d}" for a in range(1, 13)] SUBSCRIBERS = {} for i in range(200): r = rng(SEED + 37 * i) SUBSCRIBERS[f"A{i + 1:03d}"] = {"region": REGION[int(r() * 5)], "type": TYPE[int(r() * 2)], "start": 10000 + int(r() * 60000)} def generate_source(): records = [] for i, (sub, a) in enumerate(SUBSCRIBERS.items()): e = a["start"] for p in PERIODS: r = rng(SEED + 101 * i + 7 * PERIODS.index(p)) e += 4 + int(r() * 46) records.append({"subscriber": sub, "period": p, "region": a["region"], "type": a["type"], "index": e, "unit": "m3"}) return records # ---- PIPELINE: five stages, each counts the rows it drops def s1_screen(records): REQUIRED = ("subscriber", "period", "region", "index", "unit") keep = [k for k in records if all(a in k for a in REQUIRED)] return keep, len(records) - len(keep) def s2_dedupe(records): seen, keep = set(), [] for k in records: key = (k["subscriber"], k["period"]) if key in seen: continue seen.add(key) keep.append(k) return keep, len(records) - len(keep) def s3_consumption(records): table = {(k["subscriber"], k["period"]): k for k in records} out, dropped = [], 0 for k in records: i = PERIODS.index(k["period"]) if i == 0: continue prev = table.get((k["subscriber"], PERIODS[i - 1])) if prev is None: dropped += 1 continue try: f = int(k["index"]) - int(prev["index"]) except (TypeError, ValueError): dropped += 1 continue out.append(dict(k, m3=f)) return out, dropped def s4_summary(records): totals = {} for k in records: key = (k["region"], k["period"]) totals[key] = totals.get(key, 0) + k["m3"] return totals, 0 def s5_serve(summary): region = {} for (b, d), v in summary.items(): region.setdefault(b, []).append(v) return {b: round(sum(v) / len(v), 2) for b, v in region.items()}, 0 STAGES = [("screen", s1_screen), ("dedupe", s2_dedupe), ("consumption", s3_consumption), ("summary", s4_summary), ("serve", s5_serve)] def pipeline(records): trace, data = [], records for name, f in STAGES: data, dropped = f(data) trace.append({"stage": name, "remaining": len(data), "dropped": dropped}) return data, trace # ---- Of the six corruption classes of the common definition this lesson runs only one. def corrupt_schema(records, period="2026-07"): """The producer renamed the field ; the value itself stayed the same.""" out, counter = [], 0 for k in records: if k["period"] != period: out.append(k) continue y = dict(k) y["index_value"] = y.pop("index") counter += 1 out.append(y) return out, counter def add_field(records, first="2026-07"): """The producer added a harmless field ; the existing fields did not change.""" return [dict(k, meter_serial="S" + k["subscriber"][1:]) if k["period"] >= first else k for k in records] # ---- CONTRACT: the exact schema of the row the producer declares it writes CONTRACT = {"subscriber": str, "period": str, "region": str, "type": str, "index": int, "unit": str} def permissive(k): """Declared fields exist and their types are right ; an extra field is free.""" return all(a in k and isinstance(k[a], t) for a, t in CONTRACT.items()) def strict(k): """The field set must be exactly equal to the declared one ; an added field is also a violation.""" return set(k) == set(CONTRACT) and permissive(k) def ingest_with_contract(records, check=None): """The contract runs at stage 0: a conforming row passes , a non-conforming one is blocked.""" if check is None: return records, 0 passed = [k for k in records if check(k)] return passed, len(records) - len(passed) ``` ## Whom a Schema Change Breaks The second block applies six setups to the same source. The `entered` column is the rows that entered the pipeline, `blocked` the rows the contract blocked at stage 0, `screen` and `consumption` the rows those two stages dropped, and `summary` the cell count of the fourth stage. The last column says whether the five served values stayed the same as the baseline. ```python SOURCE_RECORDS = generate_source() BASELINE, _ = pipeline(SOURCE_RECORDS) def measure(name, records, check): entered, blocked = ingest_with_contract(records, check) serving, trace = pipeline(entered) print(f"{name:<28}{len(entered):>9}{blocked:>9}{trace[0]['dropped']:>8}{trace[2]['dropped']:>13}" f"{trace[3]['remaining']:>9}{'same' if serving == BASELINE else 'changed':>9}") corrupted, n = corrupt_schema(SOURCE_RECORDS) extended = add_field(SOURCE_RECORDS) print("baseline serving:", BASELINE) print("renamed rows:", n, " rows carrying the added field:", sum(1 for k in extended if "meter_serial" in k)) print() print(f"{'setup':<28}{'entered':>9}{'blocked':>9}{'screen':>8}{'consumption':>13}" f"{'summary':>9}{'serving':>9}") measure("healthy , no contract", SOURCE_RECORDS, None) measure("renamed , no contract", corrupted, None) measure("renamed , permissive", corrupted, permissive) measure("field added , no contract", extended, None) measure("field added , permissive", extended, permissive) measure("field added , strict", extended, strict) print() PRODUCER = ("subscriber", "period", "region", "type", "index_value", "unit") print("rows violating the producer's own schema:", sum(1 for k in corrupted if k["period"] == "2026-07" and not all(a in k for a in PRODUCER))) print("fields declared by the contract:", len(CONTRACT), " fields read by the pipeline:", 5) ``` ``` baseline serving: {'central': 1321.45, 'north': 1212.55, 'east': 881.36, 'west': 833.09, 'south': 1065.55} renamed rows: 200 rows carrying the added field: 1200 setup entered blocked screen consumption summary serving healthy , no contract 2400 0 0 0 55 same renamed , no contract 2400 0 200 200 45 changed renamed , permissive 2200 200 0 200 45 changed field added , no contract 2400 0 0 0 55 same field added , permissive 2400 0 0 0 55 same field added , strict 1200 1200 0 0 25 changed rows violating the producer's own schema: 0 fields declared by the contract: 6 fields read by the pipeline: 5 ``` The second to last line is the answer to the question: **the number of rows violating the producer's own schema is 0.** The producer renamed the field, wrote all the fields completely, and the write finished without error. There is no measurable flaw on its side. The side that breaks is the consumer, and it can count this only in its own pipeline. The second row gives the situation without a contract. The corrupted payload enters the pipeline as it is: the screening stage drops **200 rows** because it cannot find the required field, and the consumption stage drops **200 more rows** because it cannot find the previous value of the period that follows. That is 400 rows in total, a sixth of the source. The summary goes from 55 cells down to **45** and all five served values change. There is no error record anywhere; the pipeline completed successfully. The third row gives the situation with a contract and must be read carefully. The contract blocks 200 rows at stage 0, **detection lag 0**. But the summary is again **45** cells, and the serving has again changed. The contract did not repair the output; it only gave notice. The dropped rows stopped being scattered across stages and were gathered in one place: 200 of the 400 rows now stand at the gate rather than at the screening stage, and the remaining 200 still drop at consumption, because the previous value of the next period really does not exist. **What the contract produces is not the right number but a signal that arrives at the right time.** ## The Cost of the Contract The last three rows carry the second claim of the course. Starting from the `2026-07` period, the producer adds a harmless field to the rows: the meter serial number. None of the existing fields changes, no type drifts, and the five fields the pipeline reads stay in place. The pipeline without a contract is not affected by this change at all: 2400 rows enter, 55 cells, the serving is the same. The **permissive** contract is not affected either, because it does not count an extra field as a violation: blocked rows **0**, serving the same. The **strict** contract, however, blocks **1200 rows**. The number of rows carrying the added field is exactly 1200 as well, which means the strict contract rejects every row that sees the new field. The summary falls from 55 cells to **25**, and all five served values change. This row is not an accident; it is a consequence of the definition of the contract. Corruption: **none**. Corruption caught: **0**. Falsely blocked valid rows: **1200**. Half of the source did not enter the pipeline even though it carried no quality flaw. The cost of a check is not only the effort of writing and maintaining it; **a wrongly defined contract can stop a pipeline without any corruption at all.** The last line shows the second face of the cost. The contract declares **six** fields, and the pipeline reads **five** of them; the `type` field is not used at any stage. As the contract widens, the producer becomes tied to it, and the pipeline stops even when the type of a field the consumer never reads changes. The permissive contract's advantage over the strict one is therefore not absolute: while the permissive contract leaves the extra field free, it blocks a change in the declared fields with the same severity. The decision in writing a contract is which fields to declare, and every declared field produces both a guarantee and a lock. ## Stopping and Adapting What to do when the contract is violated is a separate decision, and its cost is measured separately. All the rows up to here are **row-level** enforcement: a row that does not conform is blocked, a row that conforms passes. The second option is **batch-level** enforcement: if there is a single non-conforming row in the payload, the whole of it is rejected. The third option is **adapting** instead of blocking: the consumer adds a stage that translates the known old name into the name in the contract. ```python def batch_level(records, check): """If even a single row does not conform , the whole payload is rejected.""" return ([], len(records)) if any(not check(k) for k in records) else (records, 0) MAPPING = {"index_value": "index"} def adapt(records): """Extra stage: translates the known old name into the name in the contract.""" out = [] for k in records: y = dict(k) for old, name in MAPPING.items(): if old in y: y[name] = y.pop(old) out.append(y) return out def report(name, entered, blocked): serving, trace = pipeline(entered) status = "same" if serving == BASELINE else ("none" if not serving else "changed") print(f"{name:<26}{len(entered):>9}{blocked:>9}{trace[3]['remaining']:>9}{status:>9}") print(f"{'enforcement':<26}{'entered':>9}{'blocked':>9}{'summary':>9}{'serving':>9}") report("row-level , permissive", *ingest_with_contract(corrupted, permissive)) report("batch-level , permissive", *batch_level(corrupted, permissive)) report("adapter + permissive", *ingest_with_contract(adapt(corrupted), permissive)) print() print("good rows falsely blocked at batch level:", sum(1 for k in corrupted if permissive(k))) print("stages added by the adapter:", 1, " names in the mapping table:", len(MAPPING)) ``` ``` enforcement entered blocked summary serving row-level , permissive 2200 200 45 changed batch-level , permissive 0 2400 0 none adapter + permissive 2400 0 55 same good rows falsely blocked at batch level: 2200 stages added by the adapter: 1 names in the mapping table: 1 ``` The three rows show three separate costs. Row-level enforcement filters out the wrong rows and keeps the pipeline running; in return it produces **an incomplete output**, and that output does not carry on itself the fact that it is incomplete. Batch-level enforcement produces no output at all; there is no serving. This is sometimes the right decision, because publishing a wrong number is more expensive than publishing no number — but its cost is plain: **2200 good rows** were falsely blocked, and all eleven healthy periods were put on hold because of the flaw of a single period. The third row is the only repair path: when the adapter stage translates the old name into the new one, all 2400 rows pass, the summary returns to **55** cells and the serving becomes **exactly the same** as the baseline. Its cost is written in two items. The pipeline is now **six-staged**, not five; and the mapping table is a burden the consumer will carry forever. As this table grows, the contract text and the schema the pipeline actually accepts drift apart: the contract says `index`, the pipeline accepts both names. If the producer one day brings the old name back with a different meaning, the mapping silently reads the wrong field, and by the definition of this course that is precisely a silent corruption. ## Summary - A data contract declares three things at once: which fields exist, what their types are, and what will happen when an undeclared field arrives. When the third is not written, the cost comes from the strict reading. - When a field name changes, **the number of rows violating the producer's own schema is 0**; the side that breaks is the consumer, and it can count the loss of 400 rows only in its own pipeline. - Without a contract the corruption is silent: 200 rows drop at screening, 200 at consumption, the summary goes from 55 cells to 45, the five served values change and no error record is produced. - Once the contract is written, the same corruption is caught **at stage 0**, but the summary is again 45 cells; the contract does not repair the output, it produces a signal at the right time. - A strict contract blocks **1200 good rows** on a harmless field addition, drops the summary to 25 cells, and does so while **corruption caught is 0**; adding a check also breaks things. - What to do on a violation is a separate cost: row-level enforcement gives an incomplete output, batch-level enforcement blocks 2200 good rows and gives no output at all, and the adapter returns the serving to the baseline but grows the pipeline to six stages and leaves the consumer a permanent mapping table. ## Next Step The contract caught the corruption at stage 0, but catching is not repairing: the 200 blocked rows have to wait somewhere, the producer has to fix them, and that period has to be given to the pipeline **again**. In no line of this lesson did the pipeline run twice. The second run is the thing the pipeline does most often, and it carries its own number: what happens to the row count when the same period is reprocessed, which stage absorbs the surplus, and does the served value show it. The next lesson runs the pipeline in two forms, **idempotent** and not, counts the duplicated rows and measures the difference in the serving.