Transformation Frameworks

The previous lesson asked where the transformation would run and put three setups side by side. All three shared a single assumption: the transformation is a chain of five stages, its order is fixed, and one person can keep it in mind. A transformation layer that accumulates at a target is not a single chain; it is dozens of separate outputs that refer to one another, and some of them never connect to the serving output at all.

This lesson writes those outputs as models. A model is a single output that has a name, and its definition is a query. Two things are measured: the tasks that dependency resolution orders and the dependencies missed by a hand-kept order. In this lesson the cost of the second shows up not as an error but as a wrong number produced without any error.

• TF9. The model registry, the query texts and the resolver are fictional; no transformation framework, query engine or data warehouse product is called. The seed is 20260218.
• TF10. A model’s dependency is derived from the ref(...) calls in its query text; it is not kept in a separate list. Writing the reference and declaring the dependency are one and the same act.
• TF11. The models are the nodes of a directed acyclic graph. The topological sorting algorithm was measured in the Graphs topic of the Data Structures course in the Computer Science curriculum and is not repeated; here the graph is built and used.
• TF12. The measured run is the first run: no table has been written before. A model that is read before its turn reads an empty table and raises no error.
• TF13. The baseline is the hand-kept order: no dependency resolution, no violation counter, no cycle check.
• TF14. The producer has re-read the commercial readings of the last period and found the indexes 12 too high. This is not a corruption but an ordinary producer correction.
• TF15. The units counted are the task, the violated dependency and the affected model. An affected model is the model that runs with missing input together with every model reachable from it.
• TF16. Resolution: models 11, declared dependencies 13, serving 5 values. No ratio is printed over the serving output, the value itself is written.

A Model’s Dependency Lives in Its Own Text

The difference between a hand-kept order list and a resolved graph is where the dependency is written. In a list, the order is a separate document: the query sits in one place, the order in another, and keeping the two consistent is a human job. In reference-based writing the dependency is inside the query: a model reading another model is the same thing as the model naming it with ref(...). The graph is derived from these names, so the moment the query changes, the graph changes with it.

The first block sets up the fictional pipeline of the previous lessons and adds ordering on top of it.

# FICTIONAL. Meter readings, the five-stage pipeline and the model registry are defined
# in the lesson; no real transformation framework, query engine or data warehouse is called.
import re

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


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


def order_tasks(nodes, edges):
    """Execution order in a directed acyclic graph."""
    incoming = {u: 0 for u in nodes}
    successors = {u: [] for u in nodes}
    for a, b in edges:
        incoming[b] += 1
        successors[a].append(b)
    ready = [u for u in nodes if incoming[u] == 0]
    order = []
    while ready:
        u = ready.pop(0)
        order.append(u)
        for v in successors[u]:
            incoming[v] -= 1
            if incoming[v] == 0:
                ready.append(v)
    return order, successors

Eleven models sit in a dictionary. correction is the newest addition: the producer has re-read the commercial readings of the last period, published the corrected indexes as a separate output, and consumption now reads it as well.

QUERY = {
    "raw_reading":       "select * from source",
    "screened":          "select * from ref('raw_reading') where required fields are filled",
    "deduplicated":      "select * from ref('screened') group by subscriber , period",
    "correction":        "select subscriber , period , new_index from ref('screened')",
    "consumption":       "select * from ref('deduplicated') left join ref('correction')",
    "region_period":     "select region , period , sum(m3) from ref('consumption')",
    "region_serving":    "select region , avg(total) from ref('region_period')",
    "type_period":       "select type , period , sum(m3) from ref('consumption')",
    "region_type":       "select * from ref('region_period') join ref('type_period')",
    "subscriber_yearly": "select subscriber , sum(m3) from ref('consumption')",
    "freshness":         "select count(*) from ref('raw_reading') , ref('region_serving')",
}
REF = re.compile(r"ref\('(\w+)'\)")


def build_graph(query):
    """Edges are derived from the references in the query text , not from a separate list."""
    nodes = list(query)
    edges = [(o, m) for m, s in query.items() for o in dict.fromkeys(REF.findall(s))]
    return nodes, edges


def violations(order, edges):
    """Which model that must run first runs later in the given order."""
    position = {u: i for i, u in enumerate(order)}
    return [(a, b) for a, b in edges if position[a] > position[b]]


def affected_models(edges, broken):
    """The model that runs with missing input: the target of the violation and everything reachable from it."""
    successors = {}
    for a, b in edges:
        successors.setdefault(a, []).append(b)
    stack = [b for _, b in broken]
    closed = set(stack)
    while stack:
        u = stack.pop()
        for v in successors.get(u, []):
            if v not in closed:
                closed.add(v)
                stack.append(v)
    return closed


NODES, EDGES = build_graph(QUERY)
ORDER, _ = order_tasks(NODES, EDGES)
WRITE_ORDER = ["raw_reading", "screened", "deduplicated", "consumption", "region_period",
               "region_serving", "freshness", "subscriber_yearly", "type_period", "region_type",
               "correction"]
# a fixed hand-kept list that has no relation to the dependencies
FIXED_LIST = ["subscriber_yearly", "screened", "region_period", "region_serving", "region_type",
              "correction", "raw_reading", "freshness", "deduplicated", "type_period", "consumption"]
ORDERS = [("resolver", ORDER), ("manual , write order", WRITE_ORDER),
          ("manual , fixed list", FIXED_LIST)]

print("model:", len(NODES), " declared dependencies:", len(EDGES))
print("resolver order:", ORDER)
print()
print(f"{'order':<24}{'model':>7}{'broken':>7}{'affected':>9}  model causing the violation")
for name, s in ORDERS:
    iv = violations(s, EDGES)
    print(f"{name:<24}{len(s):>7}{len(iv):>7}{len(affected_models(EDGES, iv)):>9}  {[a for a, _ in iv]}")

model: 11  declared dependencies: 13
resolver order: ['raw_reading', 'screened', 'deduplicated', 'correction', 'consumption', 'region_period', 'type_period', 'subscriber_yearly', 'region_serving', 'region_type', 'freshness']

order                     model broken affected  model causing the violation
resolver                     11      0        0  []
manual , write order         11      1        7  ['correction']
manual , fixed list          11      5       10  ['raw_reading', 'consumption', 'consumption', 'type_period', 'consumption']

The resolver orders the eleven models and violates none of the thirteen dependencies. The interesting part is the other two rows. The write order is the order in which the models were added to the repository, and until correction was added it was flawless: the order of ten models was correct. The new model was written to the end of the list, because new things are written there. consumption, which reads it, sits in fourth place in the list. That is a single violation, and that single violation makes seven of the eleven models run with missing input.

The fixed list is worse and more honest: it violates five dependencies and breaks ten models. Because the source read itself runs after screened, which falls to second place, the whole chain is left idle. The difference between the two rows is not the number of violations but the visibility of the violation.

What Happens When the Order Breaks

The violation count is a counter; the real question is what the run produces. The third block actually runs every order. The table of a model whose turn has not come has not been written yet, so the model that reads it reads an empty table and raises no error.

def total(records, field, paired=True):
    t = {}
    for k in records:
        key = (k[field], k["period"]) if paired else k[field]
        t[key] = t.get(key, 0) + k["m3"]
    return t


def share(region_period, type_period):
    """The region's share of the period total."""
    period_total = {}
    for (t, p), v in type_period.items():
        period_total[p] = period_total.get(p, 0) + v
    return {(b, p): round(v / period_total[p], 4) for (b, p), v in region_period.items()
            if period_total.get(p)}


# The producer re-read the last period's commercial readings: the index came out 12 higher.
FUNCS = {
    "raw_reading": lambda d: generate_source(),
    "screened": lambda d: s1_screen(d["raw_reading"])[0],
    "deduplicated": lambda d: s2_dedupe(d["screened"])[0],
    "correction": lambda d: {(k["subscriber"], k["period"]): k["index"] + 12
                             for k in d["screened"]
                             if k["period"] == "2026-12" and k["type"] == "commercial"},
    "consumption": lambda d: s3_consumption([dict(k, index=d["correction"].get(
        (k["subscriber"], k["period"]), k["index"])) for k in d["deduplicated"]])[0],
    "region_period": lambda d: s4_summary(d["consumption"])[0],
    "region_serving": lambda d: s5_serve(d["region_period"])[0],
    "type_period": lambda d: total(d["consumption"], "type"),
    "subscriber_yearly": lambda d: total(d["consumption"], "subscriber", paired=False),
    "region_type": lambda d: share(d["region_period"], d["type_period"]),
    "freshness": lambda d: {"raw": len(d["raw_reading"]), "region": len(d["region_serving"])},
}
LISTS = ("raw_reading", "screened", "deduplicated", "consumption")
EMPTY = {name: ([] if name in LISTS else {}) for name in QUERY}


def run(order):
    """First run: a model read before its turn reads an empty table."""
    store = dict(EMPTY)
    for name in order:
        store[name] = FUNCS[name](store)
    return store


for name, s in ORDERS:
    print(f"{name:<24}{run(s)['region_serving']}")

resolver                {'central': 1349.82, 'north': 1232.18, 'east': 898.82, 'west': 853.82, 'south': 1090.64}
manual , write order    {'central': 1321.45, 'north': 1212.55, 'east': 881.36, 'west': 833.09, 'south': 1065.55}
manual , fixed list     {}

The third row is the expected failure: the fixed list produces no serving output at all. An empty dictionary misleads no one; the first person to look sees that something went wrong.

The second row is the real finding of the lesson. The write order produces a serving output that is complete for all five regions. The values are in a plausible range, the relations between regions are right, no field is empty, and the run completed without error. The only problem is that the numbers are wrong: because the correction table is written after consumption, the corrected indexes never enter the calculation. The difference from the correct serving output is between 17.46 and 28.37 in every region. This is the form the course’s third claim takes in this lesson: looking at the output is not enough. The wrong order did not crash, did not warn, did not leave a gap; it silently used the old data.

The source of the difference is a single model, but its effect is not limited to a single model. The number of affected models is seven: because consumption ran with missing input, the region summary, the region serving output, the type summary, the region-type share, the subscriber yearly and the freshness report that are fed from it also carry the old number. The cost of missing one dependency is everything that sits below the missed dependency.

Two Things the Resolver Cannot See

The resolver derives the dependency from the query, so it cannot know more than the query says. Two cases show the limit: the case where the query says too much and the case where it says too little.

CYCLIC = dict(QUERY, correction=QUERY["correction"] + " where region in ref('region_serving')")
D2, K2 = build_graph(CYCLIC)
S2, _ = order_tasks(D2, K2)
print("cyclic record  model:", len(D2), " dependencies:", len(K2))
print("resolver       ordered:", len(S2), " unordered:", len(D2) - len(S2),
      " ordered tasks:", S2)
print("manual order   running models:", len(WRITE_ORDER),
      " broken edges:", len(violations(WRITE_ORDER, K2)))
print()

ACTUAL = EDGES + [("region_serving", "subscriber_yearly")]
print("declared dependencies:", len(EDGES), " actual dependencies:", len(ACTUAL))
print("violation reported by the resolver:", len(violations(ORDER, EDGES)),
      " actual violation:", len(violations(ORDER, ACTUAL)),
      " affected model:", len(affected_models(ACTUAL, violations(ORDER, ACTUAL))))

cyclic record  model: 11  dependencies: 14
resolver       ordered: 3  unordered: 8  ordered tasks: ['raw_reading', 'screened', 'deduplicated']
manual order   running models: 11  broken edges: 1

declared dependencies: 13  actual dependencies: 14
violation reported by the resolver: 0  actual violation: 1  affected model: 1

In the first case, a condition that looks at the published region serving output has been added to the correction model. This is an ordinary request and a single reference, but it puts a cycle into the graph: the region serving output comes from consumption, consumption comes from the correction, and the correction comes from the region serving output. The resolver can order only three of the eleven models, eight cannot be ordered, and the run never starts. The hand-kept order, on the other hand, runs all eleven models and passes over one edge of the cycle as a broken edge count of 1, writing it nowhere. A linear list cannot show a cycle as a cycle; it merely picks one edge of the cycle and breaks it silently. The resolver’s refusal to run is not a failure; it is the only visible proof of the cycle.

The second case is the resolver’s own cost. If a model reads another model’s output by the physical name of the table instead of ref(...), that read does not appear as a reference in the query text. The actual dependencies are fourteen, the declared ones thirteen. The resolver orders its own graph flawlessly and reports 0 violations; the actual violation is 1 and the affected model is 1. The guarantee the resolver produces rests not on the correctness of the graph but on the completeness of the declaration. Every undeclared read hides under a green report.

Summary

• A model’s dependency lives not in a separate order list but in the ref(...) calls in its own query text; because the graph is derived from that text, the graph changes when the query changes.
• Over eleven models and thirteen dependencies the resolver gives 0 violations; the write order violates 1 and the fixed list 5 dependencies, and they run 7 and 10 models with missing input respectively.
• A single model added later is enough to break the hand-kept order, because the new model is written to the end of the list but sits higher up in the graph.
• A violating run raises no error: it produces a complete serving output for five regions and deviates from the correct serving output by 17.46 to 28.37 in every region. The empty fixed-list run is less dangerous than the run that produces a wrong number.
• Adding a single reference to a query can put a cycle into the graph: the resolver orders 3 of the 11 models and does not start the run, while the manual order runs all eleven and silently breaks one edge of the cycle.
• The resolver’s guarantee rests on the completeness of the declaration: a single undeclared read raises the actual dependencies to 14, the resolver still reports 0 violations, and the actual violation is 1.

Next Step

The eleven models of this lesson sat on a single plane: all equally durable, all equally queryable. In a real transformation layer these models are split into layers, and each layer makes a different promise. The raw layer keeps the source as it is, the refined layer applies the rules, and the serving layer carries only what will be published. The cost of this separation is that the same data is stored more than once; in return, a corruption can be stopped before it travels downstream. The next lesson counts the rows each layer keeps and distributes the six corruption classes of the previous course over these three layers to measure in which layer each of them can be stopped.