Data Warehouse

The Introduction to Data Engineering course built a pipeline in ten lessons and counted it. The 2400 meter readings of the fictional measurement network passed through five stages, 2200 rows remained at the output of the consumption stage, the summary shrank to 55 cells and the serving to 5 values; then the number of stages each of the six corruption classes needed before it was caught was measured. All of this was about the rows the pipeline carries. None of it asked where those 2200 rows sit, in which layout they are written, and what that layout makes a query pay.

The question is not idle: the same 2200 rows can be put on disk in very different shapes, and the shape changes a query’s cost many times over. This lesson makes that cost measurable. It builds a storage layout, counts what a query pays to touch it, and writes down the first layout decision, enforcing the schema on write, from the side it adds and the side it costs. The course’s rule comes from here and runs for twelve lessons: the number of a storage layout is not the rows it stores but what a query pays to touch it; a layout whose query is not written down is counted as unmeasured.

• SA1. The code is a fiction; no real warehouse product, file format, catalog or query engine is called, and no product, vendor or format name is written. Approaches are named by class; SQL is a language and is free. The source and the pipeline are taken over from the Introduction to Data Engineering course, seed 20260218.
• SA2. The source is 200 subscribers x 12 periods = 2400 readings; the pipeline outputs 2200 rows, because each subscriber’s first period has no consumption difference by definition.
• SA3. A layout is which file a row falls into. write distributes rows by the partition key; if the key is empty, a single file results.
• SA4. The cost of a query is three numbers: files touched, cells read and cost = files x 120 + cells. If the filter is on the partition key, files are pruned; otherwise every file is touched.
• SA5. The per-file constant of 120 is a choice; every result sensitive to it is written together with the value of the constant.
• SA6. The baseline is the unpartitioned, row-oriented, history-free layout; only the first of these three axes is varied in this lesson.
• SA7. The six corruption classes are taken from the Introduction to Data Engineering course as input, not derived again: late_arrival, duplicated, schema_changed, unit_changed, missing_partition, type_drift. None of them is an attack; all are producer-side accidents.
• SA8. The procedure of partitioning, indexing and query planning is not repeated; it was measured in the Data Modeling and Relational Theory and Query Optimization courses. Only cost at the file level is added.
• SA9. Resolution: on 2200 rows the step of the row share is 0.0005, on at most 110 files the step of the file share is 0.0091. Cost is not a share; the value itself is written.

Layout, Files and Cost

A data warehouse is the storage class for schema-conforming structured data. Two decisions define it. The first is schema enforcement on write: to enter the warehouse, a row must fit its fields and types to a schema declared in advance. The second is that the layout is chosen in advance: which file a row falls into is decided during the write, not during the read. The second decision is measurable, because the cost of a query follows directly from it. The code below builds the pipeline, lowers its output into a layout and measures two queries on three layouts. Both select the same two columns; query A filters on period, query B on type.

# FICTIONAL storage layout -- no real warehouse product, file format or query
# engine is called. Source: FICTIONAL meter readings, seed 20260218.
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):
    """A row that lacks a required field is dropped."""
    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):
    """Consumption from the period difference. Without a previous period the row cannot be computed."""
    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


# ---- STORAGE LAYOUT: write , query , cost
COLUMNS = ("subscriber", "period", "region", "type", "index", "unit", "m3")
FILE_CONSTANT = 120        # fixed cost of touching one file , in cells


def rows():
    v, _ = s1_screen(generate_source())
    v, _ = s2_dedupe(v)
    v, _ = s3_consumption(v)
    return v


def write(records, partitioning=()):
    """Distributes the records into files by the partition key."""
    layout = {}
    for k in records:
        key = tuple(k[a] for a in partitioning)
        layout.setdefault(key, []).append(k)
    return layout


def query(layout, partitioning, selected, filter_=None, columnar=True):
    """Filter: {field: value}. Fields in the partition key prune files ,
    other fields make the whole file be read."""
    filter_ = {} if filter_ is None else filter_
    pruning = {a: v for a, v in filter_.items() if a in partitioning}
    remaining = {a: v for a, v in filter_.items() if a not in partitioning}
    files, cells, matched = 0, 0, 0
    columns_read = len(selected) if columnar else len(COLUMNS)
    for key, record in layout.items():
        if pruning:
            match = dict(zip(partitioning, key))
            if any(match[a] != v for a, v in pruning.items()):
                continue
        files += 1
        cells += len(record) * columns_read
        matched += sum(1 for k in record if all(k[a] == v for a, v in remaining.items()))
    return {"files": files, "cells": cells, "matching_rows": matched,
            "cost": files * FILE_CONSTANT + cells}


records = rows()
baseline = write(records, ())
partitioned = write(records, ("period",))
SELECTED, MEASURED = ("region", "m3"), {}
print("pipeline output rows", len(records), " selected columns", len(SELECTED), "/", len(COLUMNS))
print("layout                 files   avg.rows  A.files  A.cells     A.cost     B.cost")
for name, d, b, col in [("unpartitioned/row-wise", baseline, (), False),
                        ("unpartitioned", baseline, (), True),
                        ("period", partitioned, ("period",), True)]:
    a = query(d, b, SELECTED, {"period": "2026-07"}, columnar=col)
    v = query(d, b, SELECTED, {"type": "commercial"}, columnar=col)
    MEASURED[name] = (a, v)
    print(f"{name:<22}{len(d):>6}{len(records) / len(d):>11.1f}"
          f"{a['files']:>9}{a['cells']:>9}{a['cost']:>11}{v['cost']:>11}")

pipeline output rows 2200  selected columns 2 / 7
layout                 files   avg.rows  A.files  A.cells     A.cost     B.cost
unpartitioned/row-wise     1     2200.0        1    15400      15520      15520
unpartitioned              1     2200.0        1     4400       4520       4520
period                    11      200.0        1      400        520       5720

The table sets up the course’s triple: from here on the same three numbers stand side by side in every lesson. The layout-free baseline is the first row: one file, row-oriented read, 15400 cells for query A and 15520 cost. The layout setup is the last row: eleven files by the period key, 400 cells for query A and 520 cost. The cost in this lesson has two items: the cost of query B and the rejected rows that will be measured shortly.

The middle row separates two axes. Reading only the two selected columns brings the cells from 15400 down to 4400; this is a gain of the read format, measured separately in a later lesson. Partitioning is a different axis: the whole drop from 4400 to 400 comes from partition pruning. Because the filter is on the period field and the layout is built on the period key, the query never opens ten of the eleven files.

The last column is the first appearance of the course’s second claim: a layout also corrupts. The same period layout gives 5720 for query B, whose filter is on type; the unpartitioned baseline finishes it at 4520. Partitioning cannot prune any file here, because type is not in the key; the cells read are 4400 in both layouts, but each of the eleven files is touched separately. A layout gains for one query and loses for another, and both results stand in the same table. The course’s rule says exactly this: a layout whose query is not written down is counted as unmeasured.

The Row That Schema Enforcement Rejects

When the schema is enforced on write, the warehouse gains a gate: a row that does not conform never enters the store. What it adds and what it costs becomes visible when the six corruption classes are driven at it.

# ---- FICTIONAL corruption classes. None of them is an attack; all are
# producer-side accidents that happen in everyday operation.
CORRUPTIONS = ["late_arrival", "duplicated", "schema_changed", "unit_changed", "missing_partition", "type_drift"]


def corrupt(records, cls, period="2026-07"):
    """Corrupts a single period , the remaining periods stay intact."""
    out, counter = [], 0
    for k in records:
        if k["period"] != period:
            out.append(k)
            continue
        if cls == "late_arrival":
            if k["subscriber"] <= "A040":           # 40 subscribers never arrived this period
                counter += 1
                continue
            out.append(k)
        elif cls == "duplicated":
            out.append(k)
            if k["subscriber"] <= "A060":           # 60 records arrived twice
                out.append(dict(k))
                counter += 1
        elif cls == "schema_changed":
            y = dict(k)
            y["index_value"] = y.pop("index")       # field name changed
            counter += 1
            out.append(y)
        elif cls == "unit_changed":
            y = dict(k)
            y["index"] = k["index"] * 1000          # liters instead of m3
            y["unit"] = "m3"                        # LABEL DID NOT CHANGE , silent corruption
            counter += 1
            out.append(y)
        elif cls == "missing_partition":
            if k["region"] == "north":              # one region's partition was never written
                counter += 1
                continue
            out.append(k)
        elif cls == "type_drift":
            y = dict(k)
            y["index"] = str(k["index"])            # number turned into text
            counter += 1
            out.append(y)
    return out, counter


# ---- SCHEMA IS ENFORCED ON WRITE: a row that does not conform is never written
SCHEMA = {"subscriber": str, "period": str, "region": str, "type": str, "index": int, "unit": str}


def schema_violation(k):
    for a, t in SCHEMA.items():
        if a not in k:
            return "missing_field"
        if not isinstance(k[a], t):
            return "wrong_type"
    return None


def write_with_schema(records):
    accepted, rejected = [], {}
    for k in records:
        n = schema_violation(k)
        if n:
            rejected[n] = rejected.get(n, 0) + 1
            continue
        accepted.append(k)
    return accepted, rejected


def measurement_recoverable(k):
    """Is the rejected row's measurement still there: renamed or turned into text."""
    d = k.get("index", k.get("index_value"))
    try:
        int(d)
        return True
    except (TypeError, ValueError):
        return False


source = generate_source()
print(f"{'class':<19}{'arrived':>7}{'corrupted':>11}{'rejected':>12}  {'reason':<15}{'written':>8}")
total_rejected, total_intact, caught = 0, 0, 0
for s in ["clean"] + CORRUPTIONS:
    incoming, n = (source, 0) if s == "clean" else corrupt(source, s)
    accepted, rejected = write_with_schema(incoming)
    r = sum(rejected.values())
    reason = "-" if not rejected else "/".join(sorted(rejected))
    total_rejected += r
    total_intact += sum(1 for k in incoming if schema_violation(k) and measurement_recoverable(k))
    caught += 1 if r else 0
    print(f"{s:<19}{len(incoming):>7}{n:>11}{r:>12}  {reason:<15}{len(accepted):>8}")
print(f"rejected total {total_rejected}  measurement intact {total_intact}  "
      f"classes caught {caught}/{len(CORRUPTIONS)}")

class              arrived  corrupted    rejected  reason          written
clean                 2400          0           0  -                  2400
late_arrival          2360         40           0  -                  2360
duplicated            2460         60           0  -                  2460
schema_changed        2400        200         200  missing_field      2200
unit_changed          2400        200           0  -                  2400
missing_partition     2354         46           0  -                  2354
type_drift            2400        200         200  wrong_type         2200
rejected total 400  measurement intact 400  classes caught 2/6

What the gate adds is read off the first row: no row is rejected in the clean source, so schema enforcement does not obstruct sound data. Two classes get caught. schema_changed renamed a field, so 200 rows are turned back as a missing field; type_drift turned a number into text, so 200 rows are turned back as a wrong type. Because the fields and types of every stored row are known, the query in the previous section can select its two columns by name and finish with 400 cells: what schema enforcement adds is the right to shrink the cells a query reads.

What it costs comes in two items. The first is coverage: four of the six classes pass the gate unaffected. late_arrival and missing_partition reduce the rows, but no nonconforming row exists; a schema cannot see what did not arrive. duplicated writes a conforming row a second time, because a schema asks about fields and types, not uniqueness. unit_changed is the most dangerous: it multiplies the value by a thousand and leaves the unit label as m3, so it conforms to the schema perfectly.

The second item is the counterpart of the course’s second claim: a layout also corrupts. In all 400 of the 400 rejected rows the measurement stands intact: under the name index_value in the schema_changed rows, as text in the type_drift rows, convertible to an integer in both. The gate throws away the row, not the number. Here a layout decision does not raise data quality; it destroys recoverable data, silently, because the rejected-row count appears in no query unless written down separately.

The Per-File Constant Is a Choice

The number 120 in the cost formula is not a measured quantity but a choice of the model: it stands for the cost, in cells, of touching each file. Being a choice, its effect on each derived result must be written down separately.

def cost(r, constant):
    return r["files"] * constant + r["cells"]


(A_b, B_b), (A_p, B_p) = MEASURED["unpartitioned"], MEASURED["period"]
print(" const   A.base   A.layout    A.gain   B.base   B.layout    B.gain")
for s in (0, 120):
    print(f"{s:>6}{cost(A_b, s):>9}{cost(A_p, s):>11}{cost(A_b, s) - cost(A_p, s):>10}"
          f"{cost(B_b, s):>9}{cost(B_p, s):>11}{cost(B_b, s) - cost(B_p, s):>10}")
print("touched files difference  A", A_p["files"] - A_b["files"], " B", B_p["files"] - B_b["files"])

 const   A.base   A.layout    A.gain   B.base   B.layout    B.gain
     0     4400        400      4000     4400       4400         0
   120     4520        520      4000     4520       5720     -1200
touched files difference  A 0  B 10

The two rows separate the results cleanly. The 4000 cells that partitioning gains for query A are independent of the constant: the gain is the same at zero and at one hundred twenty, because both layouts touch a single file and the difference comes entirely from the cells read. The loss of 1200 units on query B is, by contrast, entirely the constant’s doing: at zero there is no loss, and at one hundred twenty the loss is ten files times one hundred twenty. The “a layout also corrupts” finding depends on the constant being positive, and its size is proportional to it. Sweeping the constant over several values to see which results change order is measured in a later lesson of the course.

No number in the table depends on a run. The pipeline is deterministic, random is not used, and every cell can be derived without running anything: the rows leaving the pipeline are 200 subscribers x 11 periods = 2200; the period layout has one file per period with computable consumption, 11, so 2200 / 11 = 200 rows per file; an unpartitioned read of two columns is 2200 x 2 = 4400 cells; a pruning query reads 200 x 2 = 400. Derivability is what makes the measure sufficient: later differences between layouts come from the layout decision, not from run-to-run variation.

Summary

• A storage layout is which file a row falls into; its cost is three numbers: files touched, cells read and cost = files x 120 + cells. The per-file constant of 120 is a choice.
• The same query gives 15400 cells and 15520 cost on the unpartitioned, row-oriented baseline, 4400 cells and 4520 cost when only the selected columns are read, and 400 cells and 520 cost under period partitioning.
• The same period layout gives 5720 instead of 4520 for query B, whose filter is outside the key: a layout gains for one query and loses for another, and this loss comes entirely from the per-file constant.
• Schema enforcement on write catches two of the six corruption classes and rejects 400 rows; late_arrival, duplicated, missing_partition and unit_changed pass unaffected.
• In all 400 of the 400 rejected rows the measurement stays intact: the schema gate throws away the row, not the number.
• The course’s rule: the number of a layout is not the rows it stores but what a query pays to touch it; a layout whose query is not written down is counted as unmeasured.

Next Step

The warehouse in this lesson held the gate tight and in return gave the query a cheap read. The cost of the gate was measured as well: four corruption classes pass anyway, and the 400 rejected rows never enter the store even though their measurement is recoverable. A natural question follows: what if the gate had never been built, and every incoming row were written as it is, with no schema enforced. What would the write side gain, and what would the read side pay? The next lesson measures this approach, the data lake, by the contrast of two numbers: how many cells the same query reads on the raw layout, and how many on the schema-conforming one. The second half of the table will be more uncomfortable, because with the gate removed, the corruption class that is never caught at write time is counted as well.