Write–Audit–Publish and the Boundary Contract

48.1 Why Validate at the Boundary: Write–Audit–Publish (20 min)

Imagine StatCan publishes an LFS table in which, due to a processing error, one province's unemployment rate is 64.0 instead of 6.4. With today's pipeline:

StatCan ──► raw_dlt (replaced!) ──► dbt rebuilds 20 models ──► marts ──► GX flags it ──► …too late
                                                                         dashboards already updated

The last good LFS data is gone from raw_dlt (the table was replaced), and every dependent model now contains the error.

The Write–Audit–Publish (WAP) pattern

WAP is a widely used data-engineering pattern for exactly this problem:

            WRITE                    AUDIT                         PUBLISH
StatCan ──► raw_incoming.<table> ──► GX boundary checkpoint ──┬──► raw_dlt.<table>   (pass / warnings)
                                                              │
                                                              └──► raw_quarantine.<table>__<ts>  (critical failure)
                                                                   + alert; raw_dlt keeps last good data

Properties this gives you:

• The warehouse stays consistent. A bad release can't partially overwrite good data.
• Downstream keeps working. dbt builds from the last good release, slightly stale but correct, and freshness warnings (Day 36) say so honestly.
• Evidence is preserved. The quarantined table shows exactly what StatCan sent.

Economist's analogy: a journal's peer review. Submissions (write) are reviewed (audit) before appearing in the published issue (publish); rejected ones are archived, not printed. Readers only ever see reviewed work.

48.2 Adding the Labour Force Survey Table (25 min)

The course notebook (Cell 4) reads table 14-10-0287-01, Labour force characteristics by province, monthly. Its structure:

Notice Gender vs Sex. The notebook handles both with an if "Gender" in lfs.columns … elif "Sex" … branch. That's a real schema change StatCan made to its tables. Our boundary contract will make such a change visible and deliberate instead of silently patched.

The table is large (the notebook reports over five million rows) because it crosses every dimension. That's fine for DuckDB, and a realistic test for your pipeline.

Configuration

In .dlt/config.toml, extend the full-table list from Day 30:

[sources.statcan_tables]
tables = { "18100004" = "statcan_cpi_full", "36100104" = "statcan_gdp_full", "14100287" = "statcan_lfs_full" }

Take full tables out of the direct path

From today, full tables go through WAP. ingest.py keeps loading API series (BoC, FRED, StatCan vectors) directly, but skips statcan_tables:

python ingest.py --only boc,fred,statcan

We'll update pipeline.sh accordingly on Day 50.

48.3 The Write Step: Loading into raw_incoming (25 min)

Create ingest_statcan_tables.py at the repository root (runs in .venv). Today it only writes; Day 50 adds audit and publish.

"""ingest_statcan_tables.py — Write–Audit–Publish for full StatCan tables.

Day 48: WRITE step only (load changed tables into raw_incoming).
Usage:  python ingest_statcan_tables.py [--full-refresh]
"""
import os
import sys

import dlt
import duckdb

from macro_ingest.statcan_tables import current_release_times, statcan_tables_source

DB = os.environ.get("MACRO_DUCKDB_PATH", "macro.duckdb")
FULL_REFRESH = "--full-refresh" in sys.argv


def published_release_times():
    """Last *published* releaseTime per product, from raw_dlt.statcan_releases."""
    try:
        with duckdb.connect(DB, read_only=True) as con:
            rows = con.sql("""select product_id, max(release_time)
                              from raw_dlt.statcan_releases group by 1""").fetchall()
        return {str(pid): rt for pid, rt in rows}
    except Exception:
        return {}


def changed_tables():
    tables = dlt.config["sources.statcan_tables.tables"]
    current = current_release_times(tables.keys())
    published = published_release_times()
    changed = {pid: name for pid, name in tables.items()
               if FULL_REFRESH or current.get(pid) != published.get(pid)}
    return changed, current


def write_incoming(changed):
    """WRITE: load changed tables into the holding schema raw_incoming."""
    pipeline = dlt.pipeline(pipeline_name="statcan_incoming",
                            destination="duckdb", dataset_name="raw_incoming")
    info = pipeline.run(statcan_tables_source(tables=changed))
    info.raise_on_failed_jobs()


def main():
    changed, current = changed_tables()
    if not changed:
        print("✓ No StatCan table republished since last publish — nothing to do.")
        return 0
    print("Changed tables:", changed)
    write_incoming(changed)
    print("✓ Written to raw_incoming:", ", ".join(changed.values()))
    return 0


if __name__ == "__main__":
    sys.exit(main())

Notes:

• A separate dlt pipeline (statcan_incoming) and dataset (raw_incoming) keep holding-area state apart from production (macro_ingest → raw_dlt).
• The comparison now uses the last published release, not the last loaded one. A release that was downloaded but quarantined counts as "not yet published", so it's retried when StatCan republishes.
• MACRO_DUCKDB_PATH is the same variable dbt uses (Day 31), so one setting points every tool at the warehouse.

Run it (the first run downloads all three tables, including the large LFS file):

source .venv/bin/activate
python ingest_statcan_tables.py --full-refresh
duckdb macro.duckdb -readonly -c "select table_name from information_schema.tables where table_schema = 'raw_incoming'"

48.4 Profiling the Raw LFS Table (23 min)

Profile before writing any contract (Day 45). Everything in raw_incoming is text (Day 18's raw-layer rule).

duckdb macro.duckdb -readonly

.mode box
DESCRIBE raw_incoming.statcan_lfs_full;
SELECT count(*) FROM raw_incoming.statcan_lfs_full;

-- The dimensions and their members
SELECT labour_force_characteristics, uom, count(*) AS n
FROM raw_incoming.statcan_lfs_full GROUP BY ALL ORDER BY 1;

SELECT statistics, count(*) FROM raw_incoming.statcan_lfs_full GROUP BY 1;
SELECT data_type,  count(*) FROM raw_incoming.statcan_lfs_full GROUP BY 1;
SELECT gender,     count(*) FROM raw_incoming.statcan_lfs_full GROUP BY 1;   -- or `sex` in older data
SELECT age_group,  count(*) FROM raw_incoming.statcan_lfs_full GROUP BY 1 ORDER BY 1;

-- Quality flags, and how they relate to missing values
SELECT status, value IS NULL AS value_missing, count(*) AS n
FROM raw_incoming.statcan_lfs_full GROUP BY ALL ORDER BY n DESC;

-- Period coverage
SELECT min(ref_date), max(ref_date), count(DISTINCT ref_date) FROM raw_incoming.statcan_lfs_full;

-- Is (vector, ref_date) a key?
SELECT count(*) AS rows, count(DISTINCT (vector, ref_date)) AS distinct_keys
FROM raw_incoming.statcan_lfs_full;

Record your findings in quality/specs/incoming/LFS_PROFILE_NOTES.md: exact member names, the status codes and whether they coincide with missing values, the first month, the number of rows, and whether (vector, ref_date) is unique. Every threshold tomorrow depends on these notes. Column names above assume dlt's snake_case normalisation (Day 24). Adjust to what DESCRIBE shows.

Try it 48.4

Answer with SQL: (a) Which characteristics are rates (UOM a percentage)? (b) What's the smallest labour force (in thousands) among Estimate rows, and for which province/age group? Why does that matter for checking rounding-based identities on Day 49?

-- (a)
SELECT DISTINCT labour_force_characteristics, uom
FROM raw_incoming.statcan_lfs_full
WHERE uom ILIKE '%percent%' OR uom ILIKE '%rate%';

-- (b)
SELECT geo, age_group, gender, ref_date, try_cast(value AS DOUBLE) AS lf_thousands
FROM raw_incoming.statcan_lfs_full
WHERE labour_force_characteristics = 'Labour force' AND statistics = 'Estimate'
  AND try_cast(value AS DOUBLE) IS NOT NULL
ORDER BY lf_thousands
LIMIT 5;

48.5 The Boundary Contract Spec (20 min)

The boundary contract asks: did StatCan deliver a well-formed table of the kind we expect? It checks structure, formats and keys, not economics yet.

Create the folder quality/specs/incoming/ and the file quality/specs/incoming/lfs_boundary.yml:

description: >
  Boundary contract for StatCan table 14-10-0287-01 (LFS) as delivered, before publication.
  All columns are text in the raw layer. Structural checks only.
asset: incoming_statcan_lfs_full
batch_definition: full
checkpoint_group: incoming_lfs
owner: macro-team

critical:
  - type: ExpectTableColumnsToMatchSet
    kwargs:
      column_set: [ref_date, geo, labour_force_characteristics, gender, age_group,
                   statistics, data_type, uom, vector, value, status]
      exact_match: false
    meta:
      plan_id: B01
      dimension: validity
      rationale: >
        Columns our staging model depends on. If StatCan renames a dimension (e.g. Gender ↔ Sex),
        the release is quarantined until staging is deliberately updated.

  - type: ExpectColumnValuesToNotBeNull
    kwargs: {column: ref_date}
    meta: {plan_id: B02, dimension: completeness}

  - type: ExpectColumnValueLengthsToEqual
    kwargs: {column: ref_date, value: 7}
    meta: {plan_id: B03, dimension: validity, rationale: "Monthly reference periods are 'YYYY-MM'."}

  - type: ExpectCompoundColumnsToBeUnique
    kwargs: {column_list: [vector, ref_date]}
    meta: {plan_id: B04, dimension: uniqueness,
           rationale: "Each series (vector) has exactly one value per month."}

  - type: ExpectColumnValuesToNotBeNull
    kwargs: {column: vector}
    meta: {plan_id: B05, dimension: completeness}

warning:
  - type: ExpectColumnValuesToBeInSet
    kwargs: {column: statistics, value_set: ["Estimate", "Standard error of estimate",
                                              "Standard error of month-to-month change",
                                              "Standard error of year-over-year change"]}
    meta: {plan_id: B06, dimension: validity,
           rationale: "Replace with the exact members from LFS_PROFILE_NOTES.md."}

  - type: ExpectTableRowCountToBeBetween
    kwargs: {min_value: 4500000}
    meta: {plan_id: B07, dimension: completeness,
           rationale: "Set to ~90% of the profiled row count; a much smaller table suggests a truncated download."}

Replace the member lists and row counts with your profile's values. The values above are illustrative placeholders.

Why these are the right boundary checks

48.6 Checkpoint Groups and a First validate_incoming.py (23 min)

Register the incoming asset

Add to quality/setup_gx.py (Day 43):

INCOMING_TABLES = [
    ("incoming_statcan_lfs_full", "raw_incoming", "statcan_lfs_full"),
    ("incoming_statcan_cpi_full", "raw_incoming", "statcan_cpi_full"),
    ("incoming_statcan_gdp_full", "raw_incoming", "statcan_gdp_full"),
]
TABLE_ASSETS += INCOMING_TABLES

First-time order matters. GX checks that a table exists when registering it. That's why raw_incoming was loaded in 48.3 before running setup. On later runs the tables always exist, because WAP keeps the last incoming copy in place.

Group checkpoints by purpose

Until now, all suites fed two checkpoints: critical and warning (Day 46). Incoming-data suites must run separately, per table, at audit time. Add an optional checkpoint_group to specs, and let the builder name checkpoints <group>_<severity>:

In quality/build_suites.py:

def load_specs():
    """Top-level specs and specs in subfolders (e.g. specs/incoming/), excluding drafts."""
    for path in sorted(SPEC_DIR.rglob("*.yml")):
        if "drafts" in path.parts:
            continue
        with open(path) as f:
            spec = yaml.safe_load(f)
        if not isinstance(spec, dict) or "asset" not in spec:
            continue
        yield path.stem, spec

and in main(), key the collected validation definitions by (group, severity):

    vds = {}                                               # (group, severity) → [definitions]
    ...
            group = spec.get("checkpoint_group", "")
            vds.setdefault((group, severity), []).append(vd)
    ...
    for (group, severity), defs in vds.items():
        name = f"{group}_{severity}" if group else severity
        context.checkpoints.add(gx.Checkpoint(
            name=name, validation_definitions=defs,
            actions=[gx.checkpoints.UpdateDataDocsAction(name="update_data_docs")],
            result_format={"result_format": "SUMMARY"}))
        print(f"✓ checkpoint '{name}' with {len(defs)} validation definition(s)")

Existing specs have no checkpoint_group, so the warehouse checkpoints keep their names (critical, warning). The LFS boundary spec produces incoming_lfs_critical and incoming_lfs_warning.

A first validate_incoming.py

Create quality/validate_incoming.py (runs in .venv-gx). It reuses the Day 46–47 functions:

"""Audit one incoming StatCan table with its checkpoint group.

Usage:  python quality/validate_incoming.py statcan_lfs_full
Exit:   0 pass · 1 critical failure · 2 warnings only
"""
import sys
from datetime import datetime, timezone

from gx_helpers import get_context
from run_checkpoints import history_rows, save_history, summarise

GROUPS = {
    "statcan_lfs_full": "incoming_lfs",
    "statcan_cpi_full": "incoming_cpi",
    "statcan_gdp_full": "incoming_gdp",
}


def main(table):
    group = GROUPS[table]
    context = get_context()
    run_time = datetime.now(timezone.utc)
    outcome, rows = {}, []

    for severity in ("critical", "warning"):
        try:
            cp = context.checkpoints.get(f"{group}_{severity}")
        except Exception:
            continue                                     # no suite of this severity yet
        result = cp.run()
        failed, lines = summarise(result)
        outcome[severity] = failed
        rows += history_rows(cp.name, result, run_time)
        print(f"{'✓' if failed == 0 else '✗'} {cp.name}: {failed} failing expectation(s)")
        print("\n".join(lines))

    if rows:
        save_history(rows, run_time)
    context.build_data_docs()

    if outcome.get("critical"):
        return 1
    if outcome.get("warning"):
        return 2
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1]))

source .venv-gx/bin/activate
python quality/setup_gx.py
python quality/build_suites.py
python quality/validate_incoming.py statcan_lfs_full; echo "exit: $?"

Day 48 checklist

• I can explain Write–Audit–Publish and why it keeps the warehouse consistent
• Full StatCan tables (CPI, GDP, LFS) load into raw_incoming, compared against the last published release
• I profiled the raw LFS table and recorded member names, status codes, coverage and keys
• A boundary contract checks required columns, formats, keys and size
• Specs can declare a checkpoint_group; incoming tables have their own checkpoints
• validate_incoming.py audits one table and returns 0 / 1 / 2

deactivate; git add . && git commit -m "Day 48: WAP write step, LFS boundary contract, incoming checkpoints" && git push