The Data Warehouse and Transformation Pipelines

The previous lesson dealt with a case where multiple systems wrote the same entity: a source system was chosen and the inconsistency between records was counted. There, every party was writing the data. In this lesson there is a party that never writes at all. The regional library network’s management unit only asks questions: which branch is filling up, which item is overdue, how membership is flowing. A question does not change the data, but it is not free either; every question becomes a query, every query becomes records read, and those records sit on the disk of the system that runs loan transactions.

This lesson measures the separation of analytical load from operational load. The network is a fictional model whose owners and budgets are separate.

The Questioning Party Brings a Countable Load

The model has four sources, each with a separate owner: the loan service, the membership system, the externally sourced catalog, and the branch systems. The management unit’s five questions read these.

IN10: the record counts, run frequencies, and freshness requirements are fictional and are not taken from any real institution. IN11: a report touching one entity is one query; the event entity is read by window, the dimension entity is scanned in full. IN14: a step that cannot find the field it reads breaks; a step whose input is not intact stops.

// warehouse/model.mjs — the regional library network's fictional model and transformation pipeline.
// Records are actually generated; the numbers in the next blocks are measured by scanning these arrays.
export const SEED = 7391, DAYS = 180, END = DAYS * 24 - 1;
export const OWNER = { loan: "loan service (in-house)", member: "membership system",
  item: "catalog (external)", branch: "branch systems" };

export function generate(t = SEED) {
  let s = t;
  const r = () => (s = (s * 48271) % 2147483647) / 2147483647, n = (k) => Math.floor(r() * k);
  const array = (k, f) => Array.from({ length: k }, (_, i) => f(i));
  return {
    branch: array(12, (i) => ({ id: i, region: `b${i % 4}` })),
    item: array(128_000, (i) => ({ id: i, type: ["book", "periodical", "audio"][n(3)], changeStamp: n(END) })),
    member: array(46_000, (i) => ({ id: i, regDay: n(DAYS), branchId: n(12), changeStamp: n(END) })),
    loan: array(240_000, (i) => ({ id: i, memberId: n(46_000), itemId: n(128_000),
      branchId: r() < 0.008 ? null : n(12), day: n(DAYS), hour: n(24) })),
  };
}

// Reports: entity -> window (days) or full scan; runs per day; largest accepted lag (hours);
// basis in the warehouse; field that determines breakage.
export const REPORTS = {
  occupancy: { reads: { loan: 1, branch: "full" }, runsPerDay: 48, freshness: 1, basis: "branch-summary", field: "loan.branchId" },
  "overdue-item": { reads: { loan: 30, item: "full" }, runsPerDay: 24, freshness: 4, basis: "enriched", field: "item.type" },
  "new-member": { reads: { member: "full" }, runsPerDay: 12, freshness: 14, basis: "member-summary", field: "member.regDay" },
  activity: { reads: { loan: 90, member: "full" }, runsPerDay: 2, freshness: 36, basis: "member-summary", field: "member.regDay" },
  turnover: { reads: { loan: 180, item: "full", branch: "full" }, runsPerDay: 1, freshness: 72, basis: "enriched", field: "item.type" },
};

// Records read and queries opened in a single run.
export function readReport(data, r) {
  let recordsRead = 0, queries = 0;
  for (const [entity, window] of Object.entries(r.reads)) {
    queries += 1;
    if (window === "full") { recordsRead += data[entity].length; continue; }
    for (const k of data[entity]) if (k.day >= DAYS - window) recordsRead += 1;
  }
  return { recordsRead, queries };
}

// The slice pulled in one cycle: from the event records within the period, from the dimension
// either everything or records whose change stamp falls within the period (IN13).
export function extract(data, period, dimension) {
  const threshold = END - period;
  const changed = (d) => dimension === "full" ? d : d.filter((k) => k.changeStamp > threshold);
  const slice = { loan: data.loan.filter((k) => k.day * 24 + k.hour > threshold),
    member: changed(data.member), item: changed(data.item), branch: data.branch };
  return { slice, recordsRead: Object.values(slice).reduce((t, d) => t + d.length, 0), queries: 4 };
}

// The pipeline's eight steps; each step declares the field it reads.
export const STEPS = [
  { name: "extract-loan", source: "loan", inputs: [], reads: ["memberId", "itemId", "branchId", "day"] },
  { name: "extract-member", source: "member", inputs: [], reads: ["regDay", "branchId"] },
  { name: "extract-item", source: "item", inputs: [], reads: ["type"] },
  { name: "extract-branch", source: "branch", inputs: [], reads: ["region"] },
  { name: "clean", inputs: ["extract-loan"], reads: ["branchId", "day"] },
  { name: "enriched", inputs: ["clean", "extract-item", "extract-branch"], reads: ["itemId", "type", "region"] },
  { name: "branch-summary", inputs: ["enriched"], reads: ["region", "day"] },
  { name: "member-summary", inputs: ["clean", "extract-member"], reads: ["memberId", "regDay"] },
];

const group = (d, f) => [...new Set(d.map(f))].map((a) => ({ a }));
const TRANSFORM = {
  clean: ([l]) => l.filter((k) => k.branchId !== null),
  enriched: ([l, i, b]) => {
    const t = new Map(i.map((k) => [k.id, k.type])), r = new Map(b.map((k) => [k.id, k.region]));
    return l.map((k) => ({ ...k, type: t.get(k.itemId) ?? "unknown", region: r.get(k.branchId) ?? "unknown" }));
  },
  "branch-summary": ([l]) => group(l, (k) => `${k.region}|${k.day}`),
  "member-summary": ([l]) => group(l, (k) => `${k.memberId}|${Math.floor(k.day / 30)}`),
};

// Run the pipeline: a step whose transform throws is "broken", a step whose input is not intact is "stopped".
export function run(slice) {
  const output = new Map(), status = new Map(), counts = new Map();
  let moved = 0;
  for (const s of STEPS) {
    if (s.inputs.some((d) => status.get(d) !== "done")) { status.set(s.name, "stopped"); continue; }
    const feed = s.source ? [slice[s.source]] : s.inputs.map((d) => output.get(d));
    const sample = feed.filter((d) => d.length > 0).map((d) => d[0]);
    try {
      for (const field of s.reads)
        if (sample.every((k) => (field in k) === false)) throw new Error(`missing field ${field}`);
      const result = s.source ? feed[0] : TRANSFORM[s.name](feed);
      const inCount = feed.reduce((t, d) => t + d.length, 0);
      output.set(s.name, result); status.set(s.name, "done");
      counts.set(s.name, { inCount, outCount: result.length });
      moved += inCount + result.length;
    } catch (e) { status.set(s.name, `broken: ${e.message}`); }
  }
  return { status, counts, moved };
}

The Warehouse Takes On the Load, in Exchange for a Lag

In one cycle the warehouse reads the sources, transforms them, and leaves behind the summaries the reports will read. A report gets its answer from the warehouse only if the lag it accepts is larger than the warehouse’s own lag. IN12: the pipeline moves 4,000 records per second; the freshness lag is the sum of the period and the run duration. IN13: incremental dimension loading assumes an index on the change stamp.

// warehouse/load.mjs — queries and records read that land on the operational system: without a warehouse and with one
import { generate, REPORTS, readReport, extract, run, DAYS } from "./model.mjs";

const data = generate();
const SPEED = 4000; // IN12: the pipeline moves 4,000 records per second
const measurements = Object.fromEntries(Object.entries(REPORTS).map(([name, r]) => [name, readReport(data, r)]));

console.log(`model: ${Object.entries(data).map(([name, d]) => `${name} ${d.length}`).join(", ")}; ${DAYS} days`);
console.log("\nreport          runs/day  queries/run  records/run  records/day  freshness (hrs)");
console.log("-------------- ---------- ------------ ------------- ----------- ---------------");
let rawRead = 0, rawQueries = 0;
for (const [name, r] of Object.entries(REPORTS)) {
  const o = measurements[name];
  rawRead += o.recordsRead * r.runsPerDay; rawQueries += o.queries * r.runsPerDay;
  console.log(`${name.padEnd(14)} ${String(r.runsPerDay).padStart(10)} ${String(o.queries).padStart(12)} ` +
    `${String(o.recordsRead).padStart(13)} ${String(o.recordsRead * r.runsPerDay).padStart(11)} ${String(r.freshness).padStart(15)}`);
}
console.log(`without warehouse, total: ${rawQueries} queries/day, ${rawRead} records read/day`);

console.log("\nperiod dimension      run (s)    lag  in whs  direct  queries/day records/day   % of raw");
console.log("------ ------------ --------- ------ ------- ------- ------------ ----------- ----------");
const remainingByPeriod = {};
for (const dimension of ["full", "incremental"]) for (const period of [24, 12, 3, 1]) {
  const c = extract(data, period, dimension);
  const duration = run(c.slice).moved / SPEED;
  const lag = period + duration / 3600, cycles = 24 / period;
  let queries = c.queries * cycles, recordsRead = c.recordsRead * cycles;
  const remaining = [];
  for (const [name, r] of Object.entries(REPORTS)) {
    if (r.freshness >= lag) continue;
    remaining.push(name);
    queries += measurements[name].queries * r.runsPerDay; recordsRead += measurements[name].recordsRead * r.runsPerDay;
  }
  remainingByPeriod[period] = remaining;
  console.log(`${String(period).padStart(6)} ${dimension.padEnd(12)} ${duration.toFixed(1).padStart(9)} ` +
    `${lag.toFixed(2).padStart(6)} ${String(5 - remaining.length).padStart(7)} ${String(remaining.length).padStart(7)} ` +
    `${String(Math.round(queries)).padStart(12)} ${String(Math.round(recordsRead)).padStart(11)} ` +
    `${`${(recordsRead / rawRead * 100).toFixed(1)}%`.padStart(10)}`);
}
console.log(`\nreports still operational at the 24-hour period: ${remainingByPeriod[24].join(", ")}`);
console.log(`reports still operational at the 3-hour period: ${remainingByPeriod[3].join(", ")}`);

model: branch 12, item 128000, member 46000, loan 240000; 180 days

report          runs/day  queries/run  records/run  records/day  freshness (hrs)
-------------- ---------- ------------ ------------- ----------- ---------------
occupancy              48            2          1326       63648               1
overdue-item           24            2        168293     4039032               4
new-member             12            1         46000      552000              14
activity                2            2        165808      331616              36
turnover                1            3        368012      368012              72
without warehouse, total: 163 queries/day, 5354308 records read/day

period dimension      run (s)    lag  in whs  direct  queries/day records/day   % of raw
------ ------------ --------- ------ ------- ------- ------------ ----------- ----------
    24 full             133.4  24.04       2       3          160     4830006      90.2%
    12 full             131.9  12.04       3       2          152     4451992      83.1%
     3 full             130.8   3.04       4       1          128     1456912      27.2%
     1 full             130.6   1.04       4       1          192     4241160      79.2%
    24 incremental        3.6  24.00       2       3          160     4656902      87.0%
    12 incremental        1.8  12.00       3       2          152     4104854      76.7%
     3 incremental        0.4   3.00       4       1          128       65392       1.2%
     1 incremental        0.1   1.00       4       1          192       65160       1.2%

reports still operational at the 24-hour period: occupancy, overdue-item, new-member
reports still operational at the 3-hour period: occupancy

The first table shows where the load sits. There are 163 queries and 5,354,308 records read per day; 4,039,032 of that, or seventy-five percent, comes from a single report. The reason is not the number of runs but the records read per run: the overdue-item list scans the catalog’s 128,000 rows on every run. The occupancy report, which runs 48 times a day, does not even reach one percent of the total.

The second table counts the share the warehouse takes on, period by period, and it draws a two-way curve. As the period shortens, the lag drops, and as the lag drops, more reports can get their answer from the warehouse: from two to four. But the cycle count also rises, and every cycle re-reads the sources. With full dimension loading, records read come to 90.2 percent at the 24-hour period, 27.2 percent at the 3-hour period, and back up to 79.2 percent at the 1-hour period. The cheapest point is not at either end, it is in the middle.

The column that actually decides is dimension. At the same period, incremental loading brings 27.2 percent down to 1.2 percent, because the catalog’s 128,000 rows are read only for their changed portion, not on every cycle. Whether the operational load drops is not decided by the warehouse’s existence but by how it reads the sources.

One report never moves to the warehouse at any period: the occupancy report accepts a lag of one hour, and the shortest period’s lag is 1.04 hours. The freshness requirement defines the path the warehouse cannot cover.

The Pipeline’s Steps and a Change in the Source Schema

The pipeline is eight steps, and each step declares the field it reads. A field change in the source schema is actually applied: the field is renamed or removed, the pipeline is run, and the broken step is counted.

// warehouse/breakage.mjs — records moved step by step, and the step a source-schema change breaks
import { generate, REPORTS, extract, run, STEPS, OWNER } from "./model.mjs";

const data = generate(), PERIOD = 3;
const { slice } = extract(data, PERIOD, "full");
const baseline = run(slice);
const lag = PERIOD + baseline.moved / 4000 / 3600;

console.log("step            input                                     in      out");
console.log("--------------- ----------------------------------- -------- --------");
for (const s of STEPS) {
  const c = baseline.counts.get(s.name);
  console.log(`${s.name.padEnd(15)} ${(s.source ?? s.inputs.join(",")).padEnd(35)} ` +
    `${String(c.inCount).padStart(8)} ${String(c.outCount).padStart(8)}`);
}
console.log(`one cycle moved ${STEPS.length} steps, ${baseline.moved} records; lag ${lag.toFixed(2)} hours`);

// Three schema changes applied: a field is renamed or removed.
const CHANGES = {
  "item.type -> item.typeCode": { entity: "item", field: "type", newName: "typeCode" },
  "loan.branchId -> loan.branch": { entity: "loan", field: "branchId", newName: "branch" },
  "member.regDay removed": { entity: "member", field: "regDay", newName: null },
};
const apply = (d, c) => ({ ...d, [c.entity]: d[c.entity].map((k) => {
  const y = { ...k }; delete y[c.field]; if (c.newName) y[c.newName] = k[c.field]; return y; }) });

console.log("\nchange                        owner                       broken stopped  intact dropped in whs broken w/o whs");
console.log("----------------------------- --------------------------- ------ ------- ------- -------------- --------------");
for (const [name, c] of Object.entries(CHANGES)) {
  const d = [...run(apply(slice, c)).status];
  const count = (f) => d.filter(([, v]) => f(v)).length;
  const broken = new Set(d.filter(([, v]) => v !== "done").map(([k]) => k));
  const dropped = Object.values(REPORTS).filter((r) => broken.has(r.basis)).length;
  const direct = Object.values(REPORTS).filter((r) => r.field === `${c.entity}.${c.field}`).length;
  console.log(`${name.padEnd(29)} ${OWNER[c.entity].padEnd(27)} ${String(count((v) => v.startsWith("broken"))).padStart(6)} ` +
    `${String(count((v) => v === "stopped")).padStart(7)} ${String(count((v) => v === "done")).padStart(7)} ` +
    `${String(dropped).padStart(14)} ${String(direct).padStart(14)}`);
}

// Edge: without a warehouse, a report connects to every source it reads; with one, only the pipeline does.
const rawEdges = Object.values(REPORTS).flatMap((r) => Object.keys(r.reads));
const uncovered = Object.values(REPORTS).filter((r) => r.freshness < lag);
const uncoveredEdges = uncovered.flatMap((r) => Object.keys(r.reads));
console.log(`\nedges: without warehouse ${rawEdges.length} (report -> source); with warehouse ${4 + (5 - uncovered.length) + uncoveredEdges.length} ` +
  `(4 pulls + ${5 - uncovered.length} warehouse reads + ${uncoveredEdges.length} uncovered direct reads)`);
console.log(`owners the management unit talks to: without warehouse ${new Set(rawEdges).size}, ` +
  `with warehouse ${new Set(["warehouse", ...uncoveredEdges]).size}`);

step            input                                     in      out
--------------- ----------------------------------- -------- --------
extract-loan    loan                                     146      146
extract-member  member                                 46000    46000
extract-item    item                                  128000   128000
extract-branch  branch                                    12       12
clean           extract-loan                             146      144
enriched        clean,extract-item,extract-branch     128156      144
branch-summary  enriched                                 144        4
member-summary  clean,extract-member                   46144      143
one cycle moved 8 steps, 523341 records; lag 3.04 hours

change                        owner                       broken stopped  intact dropped in whs broken w/o whs
----------------------------- --------------------------- ------ ------- ------- -------------- --------------
item.type -> item.typeCode    catalog (external)               1       2       5              3              2
loan.branchId -> loan.branch  loan service (in-house)          1       4       3              5              1
member.regDay removed         membership system                1       1       6              2              2

edges: without warehouse 10 (report -> source); with warehouse 10 (4 pulls + 4 warehouse reads + 2 uncovered direct reads)
owners the management unit talks to: without warehouse 4, with warehouse 3

The step table shows where the cost comes from. In the three-hour cycle, only 146 records come in from the loan side, but the total moved is 523,341. The whole difference is dimension reading: the enriched step takes in 128,156 records, because enriching 144 loan records means reading the entire catalog. The clean step passes 144 of the 146 records through; the two records with an empty branch are dropped.

The breakage table gives this course’s question. All three changes touch a single field, and all three break exactly one step of the pipeline. What differs is where the break stops. When the catalog’s field breaks, two more steps stop and three reports are left unanswered — without a warehouse, only two reports used that field. When the loan service’s field breaks, four steps stop and all five of the five reports drop — yet only one report read that field directly. In the membership field, the two counts come out equal.

The warehouse ties together reports that read the same source. The occupancy report never uses the catalog’s type field; without a warehouse it would be unaffected by that change, but with a warehouse it is affected, because the branch-summary step sits downstream of the enriched step. The compensation is that fixing it is localized: the break is a single step and belongs to a single team’s responsibility. Without a warehouse, the same fix has to be made separately, in each report’s independently written query.

The last lines count the edges. The edge count does not change: 10 without a warehouse, 10 with one. What changes is direction and ownership. The management unit talks to four owners without a warehouse and three with one; the two remaining owners are there because of the occupancy report the warehouse cannot cover. An integration point does not reduce the edge count — it changes who the coordination is with.

Summary

• Five reports impose 163 queries and 5,354,308 records read per day on the operational system; seventy-five percent of that comes from a single report, because it scans the catalog’s 128,000 rows on every run (IN10, IN11).
• As the period shortens, the reports answered from the warehouse rise from two to four, but because the cycle count grows, records read climb back from 27.2 percent to 79.2 percent; the cheapest period is not at either end, it is in the middle.
• Incremental dimension loading brings the operational load down from 27.2 percent to 1.2 percent at the same period; what decides it is not the warehouse’s existence but how it reads the sources (IN13).
• The report with a one-hour freshness requirement never moves to the warehouse at any period, because the shortest period’s lag is 1.04 hours. This is how the warehouse reports the path it cannot cover (IN12).
• The eight-step pipeline moves 523,341 records in one cycle; a single field change in the source breaks a single step, but in the loan field four steps stop downstream of the break, so all five of the five reports drop — only one report read that field directly (IN14).

Next Step

This lesson’s measurements leaned on an assumption: that the source systems’ fields are known, and that a schema change shows up as an error message. Not every system on the network is like that. There is a record system that has held years of loan history, and no one can reach its source code. The next lesson takes this up: when an unchangeable system’s call surface is wrapped, how many paths close, how many stay outside the wrapper, and how long the window lasts where two systems write together while data ownership is handed off.