Project: S04's freshness, volume, and lineage report

Description

This project closes the module. You have check_freshness() (lesson 4), check_volume() (lesson 5), and LINEAGE_MAP with its two query functions (lesson 7). One step remains: a closing script that brings this module's three new pieces together with the recap of the five row-level dimensions module 5 already closed, and reports, with executed evidence, the complete state of Kiosko's data quality system after six modules — the exact moment DIMENSIONS_STILL_OPEN, the list that had been carrying over since module 3, finally goes empty.

Connection to the module. This project introduces no new concept — it's the synthesis of lessons 2 through 7, applied end to end to the same incident the five earlier modules diagnosed. It closes the thread module 1 opened (six data quality dimensions) and the thread this module's lesson 6 opened (what lineage answers that no contract answers), leaving this guide ready for module 7: what to do, in practice, when these checks fail in production.

An analogy: the audit report, with the last two sections added

Module 3 built the complete audit report analogy, with one section per reviewed area and an explicit list of what still hadn't been reviewed. Module 5 added the accuracy section, and left the pending list at a single item: ["freshness"]. This project is that same audit, with the last two sections added — freshness, and a completely new one no earlier report in this guide had: lineage — and with the data quality dimensions pending list, for the first time since module 1, completely empty.

The material you need

You need, in this module's same working folder:

module_6_freshness_volume_lineage/
├── kiosko.duckdb                          (orders_s04 from module 2)
└── s04_trust_report.py                    (this project)

If your kiosko.duckdb doesn't have the orders_s04 table yet, repeat module 2's lesson 4, step 2 before continuing — this project doesn't re-explain that step, it assumes you've already done it. dim_product gets rebuilt inside this same script, with the same four products modules 3 and 5 already used, so this project is self-sufficient with no dependency on another script having already run.

The verified reference solution

# s04_trust_report.py -- module 6 closing project
import duckdb
import polars as pl
from datetime import datetime

PIPELINE_RUN_AT = "2026-08-16T09:00:00"

# REFERENCE_PRICES was already calculated and verified in module 5, lesson 4, over the 40
# rows of the canonical week (S01-S03). It gets reused here as an already-confirmed
# constant, exactly the way PIPELINE_RUN_AT gets reused with no re-derivation in every lesson.
REFERENCE_PRICES = {"P001": 0.55, "P002": 1.2, "P003": 0.75, "P004": 4.5}

LINEAGE_MAP = {
    "fact_orders.order_id": ["orders.order_id"],
    "fact_orders.store_id": ["orders.store_id"],
    "fact_orders.product_id": ["orders.product_id"],
    "fact_orders.quantity": ["orders.quantity"],
    "fact_orders.unit_price": ["orders.unit_price"],
    "fact_orders.revenue": ["orders.quantity", "orders.unit_price"],
    "fact_orders.order_ts": ["orders.order_ts"],
    "dim_product.product_id": ["products.product_id"],
    "dim_product.product_name": ["products.product_name"],
    "dim_product.category": ["products.category"],
    "dim_product.unit_cost": ["products.unit_cost"],
    "dim_store.store_id": ["stores.store_id"],
    "dim_store.store_name": ["stores.store_name"],
    "dim_store.city": ["stores.city"],
    "dim_store.country": ["stores.city"],
}


def check_freshness(df: pl.DataFrame, run_at: str, sla_hours: int, timestamp_col: str = "order_ts") -> dict:
    """Compares the DataFrame's most recent order_ts against run_at. Fails if it exceeds sla_hours."""
    latest_ts = df.select(pl.col(timestamp_col).max()).item()
    run_at_dt = datetime.fromisoformat(run_at)
    hours_since_latest = (run_at_dt - latest_ts).total_seconds() / 3600
    return {
        "check": "freshness",
        "latest_row_ts": str(latest_ts),
        "run_at": run_at,
        "sla_hours": sla_hours,
        "hours_since_latest": round(hours_since_latest, 2),
        "status": "PASS" if hours_since_latest <= sla_hours else "FAIL",
    }


def check_volume(df: pl.DataFrame, min_rows: int, max_rows: int) -> dict:
    """Confirms that df.height falls within [min_rows, max_rows]."""
    row_count = df.height
    return {
        "check": "volume",
        "row_count": row_count,
        "min_rows": min_rows,
        "max_rows": max_rows,
        "status": "PASS" if min_rows <= row_count <= max_rows else "FAIL",
    }


def check_price_baseline(df: pl.DataFrame, reference_prices: dict[str, float], tolerance: float = 0.5) -> pl.DataFrame:
    """Rows in df whose unit_price deviates from the baseline beyond tolerance (module 5)."""
    return (
        df.with_columns(
            pl.col("product_id").replace_strict(reference_prices, default=None).alias("reference_price")
        )
        .filter(pl.col("unit_price").is_not_null() & pl.col("reference_price").is_not_null())
        .with_columns(
            ((pl.col("unit_price") - pl.col("reference_price")).abs() / pl.col("reference_price")).alias("deviation")
        )
        .filter(pl.col("deviation") > tolerance)
    )


def validate_referential_integrity(orders_df: pl.DataFrame, dim_product_df: pl.DataFrame) -> pl.DataFrame:
    """Rows in orders_df whose product_id does NOT exist in dim_product_df (module 3)."""
    return orders_df.join(dim_product_df, on="product_id", how="anti")


def trace_column(column: str, lineage_map: dict[str, list[str]]) -> list[str]:
    """Returns a derived column's source columns, or [] if it isn't mapped."""
    return lineage_map.get(column, [])


def lineage_report(lineage_map: dict[str, list[str]]) -> pl.DataFrame:
    """Flattens LINEAGE_MAP into a table: derived column, source columns, how many."""
    rows = [
        {"derived_column": target, "source_columns": ", ".join(sources), "n_sources": len(sources)}
        for target, sources in lineage_map.items()
    ]
    return pl.DataFrame(rows).sort("derived_column")


def main() -> None:
    pl.Config.set_tbl_rows(20)
    pl.Config.set_fmt_str_lengths(40)

    con = duckdb.connect("kiosko.duckdb")
    con.execute("""
        CREATE OR REPLACE TABLE dim_product (
            product_id VARCHAR, product_name VARCHAR, category VARCHAR, unit_cost DOUBLE
        )
    """)
    con.execute("""
        INSERT INTO dim_product VALUES
            ('P001', 'Bottled Water 600ml', 'beverages', 0.40),
            ('P002', 'Energy Bar', 'snacks', 0.60),
            ('P003', 'Instant Coffee Sachet', 'beverages', 0.35),
            ('P004', 'Phone Charger Cable', 'electronics', 2.10)
    """)

    df = con.sql("SELECT * FROM orders_s04").pl()
    dim_product_df = con.sql("SELECT * FROM dim_product").pl()

    print("=== Kiosko: S04's freshness, volume, and lineage report ===\n")
    print(f"orders_s04: {df.height} rows | dim_product: {dim_product_df.height} rows\n")

    # --- Recap: the five row-level dimensions, already confirmed in M1-M5 ---
    flagged_completeness = set(df.filter(pl.col("unit_price").is_null())["order_id"].to_list())
    flagged_validity = set(df.filter(pl.col("quantity") <= 0)["order_id"].to_list())
    orphans = validate_referential_integrity(df, dim_product_df)
    flagged_consistency = set(orphans["order_id"].to_list())
    price_anomalies = check_price_baseline(df, REFERENCE_PRICES, tolerance=0.5)
    flagged_accuracy = set(price_anomalies["order_id"].to_list())
    dup_counts = df.group_by("order_id").agg(pl.len().alias("n"))
    duplicated_ids = set(dup_counts.filter(pl.col("n") > 1)["order_id"].to_list())
    flagged_uniqueness_rows = (
        df.with_row_index().filter(pl.col("order_id").is_in(list(duplicated_ids)))
        .sort("index").group_by("order_id").tail(1)
    )
    flagged_uniqueness = set(flagged_uniqueness_rows["order_id"].to_list())

    row_dimensions = {
        "completeness": flagged_completeness,
        "uniqueness": flagged_uniqueness,
        "validity": flagged_validity,
        "consistency": flagged_consistency,
        "accuracy": flagged_accuracy,
    }
    print("=== 1. Recap: the five row-level dimensions (M1-M5) ===")
    for dim, ids in row_dimensions.items():
        print(f"  {dim:<14}{sorted(ids)}")

    # --- What's new in this module: freshness, volume, lineage ---
    print("\n=== 2. check_freshness(df, run_at=PIPELINE_RUN_AT, sla_hours=24) ===")
    freshness_result = check_freshness(df, run_at=PIPELINE_RUN_AT, sla_hours=24)
    for k, v in freshness_result.items():
        print(f"  {k}: {v}")

    print("\n=== 3. check_volume(df, min_rows=5, max_rows=20) ===")
    volume_result = check_volume(df, min_rows=5, max_rows=20)
    for k, v in volume_result.items():
        print(f"  {k}: {v}")

    print("\n=== 4. LINEAGE_MAP -- complete report ===")
    print(lineage_report(LINEAGE_MAP))

    # --- The close: the six dimensions, plus volume and lineage ---
    n_row_dimensions_with_findings = sum(1 for ids in row_dimensions.values() if ids)
    dimensions_still_open: list[str] = []

    print("\n=== 5. Close: state of S04's trust system ===")
    print(f"  Row-level dimensions with at least one finding: {n_row_dimensions_with_findings} of 5")
    print(f"  Freshness: {freshness_result['status']} ({freshness_result['hours_since_latest']}h of {freshness_result['sla_hours']}h SLA)")
    print(f"  Data quality dimensions with a runnable check: 6 of 6")
    print(f"  Volume: {volume_result['status']} ({volume_result['row_count']} rows, range [{volume_result['min_rows']}, {volume_result['max_rows']}])")
    print(f"  Lineage: {len(LINEAGE_MAP)} warehouse columns mapped to their source")
    print(f"  DIMENSIONS_STILL_OPEN: {dimensions_still_open}")


if __name__ == "__main__":
    main()

What to expect (verified by actually running python3 s04_trust_report.py, with kiosko.duckdb containing orders_s04, polars==1.43.2, duckdb==1.5.5):

=== Kiosko: S04's freshness, volume, and lineage report ===

orders_s04: 12 rows | dim_product: 4 rows

=== 1. Recap: the five row-level dimensions (M1-M5) ===
  completeness  ['ORD-9503']
  uniqueness    ['ORD-9502']
  validity      ['ORD-9507']
  consistency   ['ORD-9508']
  accuracy      ['ORD-9509']

=== 2. check_freshness(df, run_at=PIPELINE_RUN_AT, sla_hours=24) ===
  check: freshness
  latest_row_ts: 2026-08-14 09:25:00
  run_at: 2026-08-16T09:00:00
  sla_hours: 24
  hours_since_latest: 47.58
  status: FAIL

=== 3. check_volume(df, min_rows=5, max_rows=20) ===
  check: volume
  row_count: 12
  min_rows: 5
  max_rows: 20
  status: PASS

=== 4. LINEAGE_MAP -- complete report ===
shape: (15, 3)
┌──────────────────────────┬────────────────────────────────────┬───────────┐
│ derived_column           ┆ source_columns                     ┆ n_sources │
│ ---                      ┆ ---                                ┆ ---       │
│ str                      ┆ str                                ┆ i64       │
╞══════════════════════════╪════════════════════════════════════╪═══════════╡
│ dim_product.category     ┆ products.category                  ┆ 1         │
│ dim_product.product_id   ┆ products.product_id                ┆ 1         │
│ dim_product.product_name ┆ products.product_name              ┆ 1         │
│ dim_product.unit_cost    ┆ products.unit_cost                 ┆ 1         │
│ dim_store.city           ┆ stores.city                        ┆ 1         │
│ dim_store.country        ┆ stores.city                        ┆ 1         │
│ dim_store.store_id       ┆ stores.store_id                    ┆ 1         │
│ dim_store.store_name     ┆ stores.store_name                  ┆ 1         │
│ fact_orders.order_id     ┆ orders.order_id                    ┆ 1         │
│ fact_orders.order_ts     ┆ orders.order_ts                    ┆ 1         │
│ fact_orders.product_id   ┆ orders.product_id                  ┆ 1         │
│ fact_orders.quantity     ┆ orders.quantity                    ┆ 1         │
│ fact_orders.revenue      ┆ orders.quantity, orders.unit_price ┆ 2         │
│ fact_orders.store_id     ┆ orders.store_id                    ┆ 1         │
│ fact_orders.unit_price   ┆ orders.unit_price                  ┆ 1         │
└──────────────────────────┴────────────────────────────────────┴───────────┘

=== 5. Close: state of S04's trust system ===
  Row-level dimensions with at least one finding: 5 of 5
  Freshness: FAIL (47.58h of 24h SLA)
  Data quality dimensions with a runnable check: 6 of 6
  Volume: PASS (12 rows, range [5, 20])
  Lineage: 15 warehouse columns mapped to their source
  DIMENSIONS_STILL_OPEN: []

Read the full result with the same care you've already trained in earlier projects. Section 1 recaps, with nothing new recalculated — the same five order_id, exactly as module 5's project left them — the five row-level dimensions that already have their own runnable check. Sections 2 and 3 are this module's new work, really run: freshness fails with 47.58 hours over a 24-hour SLA; volume passes with 12 rows within [5, 20]. Section 4 confirms, with the complete table, that Kiosko's warehouse's fifteen columns already have their lineage documented. And section 5 is the line that closes this entire guide's narrative thread: six of six data quality dimensions with a runnable check, and DIMENSIONS_STILL_OPEN: [] — empty, for the first time, since that list appeared in module 3's project.

Diagram: the complete thread, from module 1 to this project

flowchart TD
    A["Module 1: 6 dimensions\nnamed, 0 with a check"] --> B["Module 2: +3\ncompleteness, uniqueness, validity"]
    B --> C["Module 3: +1\nconsistency -- DIMENSIONS_STILL_OPEN\nappears for the first time"]
    C --> D["Module 4: contracts --\nsame 4, now versioned"]
    D --> E["Module 5: +1\naccuracy -- STILL_OPEN = ['freshness']"]
    E --> F["Module 6 (this project): +1\nfreshness -- STILL_OPEN = []"]
    F --> G["This module's bonus:\nvolume (PASSES) + lineage\n(15 columns mapped)"]
    G --> H["Module 7:\nwhat to do when this fails\nin production -- quarantine,\nalert, runbook, governance"]

Closing the module's promise, point by point

Common mistakes

Thinking DIMENSIONS_STILL_OPEN = [] means S04 is already "fixed" and ready for production. What happens: someone, seeing the pending list empty for the first time in the guide, concludes work on S04 is finished. Why it happens: an empty list feels, intuitively, like "nothing left to do." How to spot it: review exactly what that list means — it counts dimensions with an available runnable check, not rows with no problem. The same six broken rows module 1 diagnosed remain exactly as broken; S04 still has a file that arrived 47.58 hours late. How to fix it: always distinguish "we have the tools to detect every known problem" (what this project did achieve) from "there's no problem" (something this project never claimed, and the report itself explicitly contradicts with freshness: FAIL).

Recalculating the five row-level dimensions from scratch, instead of recognizing this project only adds new evidence about freshness/volume/lineage. What happens: someone, extending this project, writes new filtering logic for completeness or validity, without realizing row_dimensions already calculates them, with the exact same criterion modules 1 through 5 already verified. Why it happens: a long script, with many sections, can make you lose track of what's already calculated and available as a variable. How to spot it: if your extension to this project rewrites df.filter(pl.col("quantity") <= 0) or something equivalent, you're already duplicating work row_dimensions["validity"] already did. How to fix it: any extension of this report should build from the variables main() already calculated — row_dimensions, freshness_result, volume_result — the exact same habit module 5's project already recommended.

Confusing this project's 6 of 6 number with 12 of 12 (all the file's rows). What happens: someone reads "6 of 6 dimensions with a runnable check" and interprets it as describing how many of the file's twelve rows are okay. Why it happens: the number 6 appears twice in this guide's context — six data quality dimensions, and also six of the file's twelve physical rows with some known problem (module 5's project finding) — and it's easy to mix them up. How to spot it: check carefully what each 6 refers to — "6 of 6 dimensions" counts available tools; "6 of 12 rows" counts physical rows with some problem, a completely different number this project doesn't recalculate. How to fix it: whenever you cite any number from this guide, always check its exact unit — dimensions, physical rows, distinct order_id, mapped columns — the same discipline this module's lessons 2 and 5's common mistakes already demanded.

Exercises

Exercise 1 — Run the whole project yourself, from scratch. In a new folder, with kiosko.duckdb containing only orders_s04 (module 2), run python3 s04_trust_report.py. Confirm you see exactly 6 of 6 dimensions, freshness: FAIL, volume: PASS, and DIMENSIONS_STILL_OPEN: [].

Exercise 2 — Add a sixth "dimension" to the report: a count of how many warehouse columns still have no lineage mapping. Using LINEAGE_MAP and the complete list of columns you already know from fact_orders, dim_product, and dim_store (fifteen total, according to lesson 7), confirm with code that lineage coverage is 100%.

ALL_WAREHOUSE_COLUMNS = [
    "fact_orders.order_id", "fact_orders.store_id", "fact_orders.product_id",
    "fact_orders.quantity", "fact_orders.unit_price", "fact_orders.revenue", "fact_orders.order_ts",
    "dim_product.product_id", "dim_product.product_name", "dim_product.category", "dim_product.unit_cost",
    "dim_store.store_id", "dim_store.store_name", "dim_store.city", "dim_store.country",
]
unmapped = [col for col in ALL_WAREHOUSE_COLUMNS if col not in LINEAGE_MAP]
coverage_pct = round(100 * (len(ALL_WAREHOUSE_COLUMNS) - len(unmapped)) / len(ALL_WAREHOUSE_COLUMNS), 1)
print(f"Unmapped columns: {unmapped}")
print(f"Lineage coverage: {coverage_pct}%")

Unmapped columns: []
Lineage coverage: 100.0%

Exercise 3 — Argue whether check_volume() should run before or after check_freshness() in a real pipeline, and whether the order matters for this project's final result. In 2-3 sentences, considering both functions are completely independent (confirmed in lesson 5), argue whether the order they appear in within main() affects the result.

Summary and next step: closing this module

With this project you close module 6, and with it, the complete diagnosis of the six data quality dimensions module 1 opened. You learned why freshness is a whole-file question, never a row's (lesson 2), why the "now" reference has to be an injected constant, never the real clock (lesson 3), and you wrote check_freshness(), this entire guide's first file-level function that really runs and fails with evidence (lesson 4). You built check_volume(), with the deliberate contrast that S04 does pass that check (lesson 5). And, beyond the six dimensions, you opened a completely new question — where every piece of data comes from — that neither the contract nor any earlier tool could answer (lesson 6), and you resolved it with LINEAGE_MAP, this entire ecosystem's first complete lineage map, naming OpenLineage and Marquez as its production version (lesson 7).

With this project, S04's six data quality dimensions each have their own runnable check: completeness, uniqueness, and validity (modules 1-2), consistency (module 3), accuracy (module 5), and now freshness (this module) — and, beyond those six, S04 has a volume check (PASS) and a complete lineage map (fifteen columns). What's still missing isn't more detection — it's decision: what to do, in practice, when any of these checks fails in production, with no crashing like foundations M7 and no passing silently like the green checkmark that opened this guide.

Where you go next. Module 7 — The incident and data governance — answers exactly that: quarantine() separates good rows from bad ones, instead of rejecting the whole file; raise_alert() structures a notification, with no real call to any external system; a runbook.md documents, step by step, what to do with an incident like S04's. And that same module adds data governance — who can see which column, how to mask sensitive information deterministically, a minimal catalog of all of Kiosko's tables — closing this guide's complete mandate.

Resources

• Polars — complete official documentation (select, max, height, date expressions, group_by, join — the complete set of operations used across this module). docs.pola.rs. In English.
• Module 5, project (lesson 8), of this same guide — the source of REFERENCE_PRICES and the honest-reporting pattern (DIMENSIONS_STILL_OPEN) this project picks back up. src/guides/data-reliability-and-governance-guide/workbook/module-05-accuracy-and-deterministic-anomaly-detection/en/08-project-s04s-accuracy-audit.md. In English.
• Module 6, lesson 7, of this same guide — the exact source of LINEAGE_MAP, trace_column(), and lineage_report(). src/guides/data-reliability-and-governance-guide/workbook/module-06-freshness-volume-and-lineage/en/07-mapping-kioskos-lineage-by-hand.md. In English.
• This guide's DESIGN — the complete map of all eight modules, including the module 7 that follows. src/guides/data-reliability-and-governance-guide/DISENO.md. In Spanish.