Schema Contracts

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.

# 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.

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.

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.