The Staging Layer: CPI

18.1 What Staging Is (and Isn't) (20 min)

A staging model is a clean, faithful copy of one source table. Its job is to make the data easy and safe to use, not to answer economic questions.

Why be so strict? Because many downstream models will depend on staging. If a staging view quietly keeps only national data, the provincial analysis someone builds next month silently breaks. Keep staging general; put decisions downstream.

Grain. Every model must state what one row represents. The grain of stg_statcan__cpi is one row per month × geography × product group.

18.2 A Proper Raw Layer: Keep Everything, as Text (25 min)

On Day 13, raw.statcan_cpi already renamed and filtered columns. That's staging work done too early. A professional raw layer keeps every column exactly as delivered, read as text so nothing is mis-parsed.

Make sure you have both full tables

CPI was unzipped on Day 13. Download and unzip GDP (table 36-10-0104) with the Day 10 helper:

python - <<'EOF'
import sys
sys.path.insert(0, "day10")
from statcan_tables import download_table
import requests
download_table("36100104", requests.Session())
EOF
mkdir -p data/raw/statcan/36100104
unzip -o data/raw/statcan/36100104.zip -d data/raw/statcan/36100104
ls data/raw/statcan/*/

(python - <<'EOF' ... EOF runs a short Python snippet straight from the terminal.)

Add full raw tables

Append to sql/02_raw_tables.sql:

-- Full StatCan tables, every column, all text. Grain = source file rows.
CREATE OR REPLACE TABLE raw.statcan_cpi_full AS
SELECT * FROM read_csv('data/raw/statcan/18100004/18100004.csv',
                       header = true, all_varchar = true);

CREATE OR REPLACE TABLE raw.statcan_gdp_full AS
SELECT * FROM read_csv('data/raw/statcan/36100104/36100104.csv',
                       header = true, all_varchar = true);

Run it in the CLI:

duckdb macro.duckdb

.mode box
.read sql/02_raw_tables.sql
DESCRIBE raw.statcan_cpi_full;

You'll see columns you haven't used yet:

If DESCRIBE shows the first column with an odd prefix (a hidden byte-order mark, Day 13), note its exact name and use it in double quotes in the staging view.

18.3 Profiling the Raw CPI Table (23 min)

Before writing staging logic, profile: understand what's really in the data. Create day18/profile_cpi.sql.

-- How big, how old?
SELECT COUNT(*) AS n_rows, MIN(REF_DATE) AS first, MAX(REF_DATE) AS last
FROM raw.statcan_cpi_full;

-- Which geographies?
SELECT GEO, COUNT(*) AS n_rows
FROM raw.statcan_cpi_full
GROUP BY GEO
ORDER BY n_rows DESC;

-- Which index bases?
SELECT UOM, COUNT(*) AS n_rows
FROM raw.statcan_cpi_full
GROUP BY UOM;

-- Quality flags
SELECT STATUS, COUNT(*) AS n_rows
FROM raw.statcan_cpi_full
GROUP BY STATUS
ORDER BY n_rows DESC;

-- Values that won't convert to numbers
SELECT VALUE, COUNT(*) AS n_rows
FROM raw.statcan_cpi_full
WHERE TRY_CAST(VALUE AS DOUBLE) IS NULL
GROUP BY VALUE;

-- Discontinued series
SELECT COUNT(DISTINCT VECTOR) AS terminated_series
FROM raw.statcan_cpi_full
WHERE TERMINATED = 't';

-- How many product groups? (There are many more than the headline aggregates.)
SELECT COUNT(DISTINCT "Products and product groups") FROM raw.statcan_cpi_full;

Try it 18.3

Answer with queries: (a) Which geographies are cities rather than provinces? (Hint: city names contain a comma, e.g. 'Toronto, Ontario'.) (b) For GEO = 'Canada', how many product groups have data for the latest month? (c) Is (REF_DATE, GEO, product group) unique in the raw table?

-- (a)
SELECT DISTINCT GEO FROM raw.statcan_cpi_full WHERE GEO LIKE '%,%' ORDER BY GEO;

-- (b)
SELECT COUNT(DISTINCT "Products and product groups")
FROM raw.statcan_cpi_full
WHERE GEO = 'Canada'
  AND REF_DATE = (SELECT MAX(REF_DATE) FROM raw.statcan_cpi_full)
  AND TRY_CAST(VALUE AS DOUBLE) IS NOT NULL;

-- (c) Zero rows means unique
SELECT REF_DATE, GEO, "Products and product groups", COUNT(*) AS n
FROM raw.statcan_cpi_full
GROUP BY ALL
HAVING COUNT(*) > 1
LIMIT 5;

18.4 Seeds: Small Reference Tables (25 min)

Some information doesn't come from any API: province codes, the notebook's recession dates, CPI basket weights. dbt calls these hand-maintained tables seeds. We'll keep them in sql/03_seeds.sql, in a dedicated schema so they're easy to find.

-- sql/03_seeds.sql — hand-maintained reference tables
CREATE SCHEMA IF NOT EXISTS seeds;

-- Grain: one row per geography used in national/provincial analysis
CREATE OR REPLACE TABLE seeds.seed_geo (
    geo        VARCHAR PRIMARY KEY,
    geo_code   VARCHAR NOT NULL,
    geo_level  VARCHAR NOT NULL,
    region     VARCHAR
);
INSERT INTO seeds.seed_geo VALUES
    ('Canada',                    'CA', 'national', NULL),
    ('Newfoundland and Labrador', 'NL', 'province', 'Atlantic'),
    ('Prince Edward Island',      'PE', 'province', 'Atlantic'),
    ('Nova Scotia',               'NS', 'province', 'Atlantic'),
    ('New Brunswick',             'NB', 'province', 'Atlantic'),
    ('Quebec',                    'QC', 'province', 'Central'),
    ('Ontario',                   'ON', 'province', 'Central'),
    ('Manitoba',                  'MB', 'province', 'Prairies'),
    ('Saskatchewan',              'SK', 'province', 'Prairies'),
    ('Alberta',                   'AB', 'province', 'West'),
    ('British Columbia',          'BC', 'province', 'West');

-- Grain: one row per recession band, as defined in the course notebook
CREATE OR REPLACE TABLE seeds.seed_recession_bands (
    label       VARCHAR PRIMARY KEY,
    start_date  DATE NOT NULL,
    end_date    DATE NOT NULL,
    CHECK (end_date >= start_date)
);
INSERT INTO seeds.seed_recession_bands VALUES
    ('GFC',              '2008-10-01', '2009-06-30'),
    ('Oil shock',        '2015-01-01', '2015-09-30'),
    ('COVID',            '2020-02-01', '2020-06-30'),
    ('Tariff recession', '2025-10-01', '2026-03-31');

-- Grain: one row per major CPI component.
-- Weights (%) as used in the course notebook's Module 05. Illustrative only:
-- official basket weights are published in StatCan table 18-10-0007-01.
-- 'Energy' is omitted because it overlaps Shelter and Transportation.
CREATE OR REPLACE TABLE seeds.seed_cpi_weights (
    component  VARCHAR PRIMARY KEY,
    weight_pct DOUBLE NOT NULL CHECK (weight_pct > 0 AND weight_pct < 100)
);
INSERT INTO seeds.seed_cpi_weights VALUES
    ('Shelter',                                          29.0),
    ('Transportation',                                   16.0),
    ('Food',                                             15.9),
    ('Household operations, furnishings and equipment',  14.0),
    ('Recreation, education and reading',                10.4),
    ('Health and personal care',                          5.4),
    ('Clothing and footwear',                             5.0);

Run it with .read sql/03_seeds.sql, then add the schema to sql/01_schemas.sql:

CREATE SCHEMA IF NOT EXISTS seeds;
CREATE SCHEMA IF NOT EXISTS intermediate;

Seeds are code. Because they live in a .sql file in Git, every change to a weight or a recession date is versioned and reviewable. That's much better than a spreadsheet on someone's desktop.

Documenting tables inside the database

DuckDB stores comments with your tables so the documentation travels with the data:

COMMENT ON TABLE seeds.seed_cpi_weights IS
    'Illustrative CPI weights from the course notebook; see StatCan 18-10-0007-01 for official weights.';

SELECT schema_name, table_name, comment
FROM duckdb_tables()
WHERE comment IS NOT NULL;

18.5 Building staging.stg_statcan__cpi (20 min)

Create sql/10_stg_statcan__cpi.sql. Notice the structure, which you'll reuse for every staging model: source → renamed → final.

-- staging.stg_statcan__cpi
-- Source:  raw.statcan_cpi_full (StatCan table 18-10-0004-01, CPI, monthly, NSA)
-- Grain:   one row per month × geography × product group
-- Notes:   rows without a numeric value are dropped (not available / suppressed).
--          Terminated series are kept but flagged.

CREATE OR REPLACE VIEW staging.stg_statcan__cpi AS
WITH source AS (
    SELECT * FROM raw.statcan_cpi_full
),

renamed AS (
    SELECT
        CAST(REF_DATE || '-01' AS DATE)          AS month,
        GEO                                      AS geo,
        "Products and product groups"            AS component,
        VECTOR                                   AS vector_id,
        UOM                                      AS index_base,
        TRY_CAST(VALUE AS DOUBLE)                AS cpi,
        NULLIF(TRIM(STATUS), '')                 AS status_flag,
        COALESCE(TERMINATED = 't', false)        AS is_terminated
    FROM source
)

SELECT r.month,
       r.geo,
       g.geo_code,
       COALESCE(g.geo_level, 'city')             AS geo_level,
       r.component,
       r.vector_id,
       r.index_base,
       r.cpi,
       r.status_flag,
       r.is_terminated
FROM renamed AS r
LEFT JOIN seeds.seed_geo AS g USING (geo)
WHERE r.cpi IS NOT NULL;

COMMENT ON VIEW staging.stg_statcan__cpi IS
    'CPI (18-10-0004-01), typed. Grain: month × geo × component. NSA.';

Run and inspect:

.read sql/10_stg_statcan__cpi.sql
DESCRIBE staging.stg_statcan__cpi;

SELECT geo_level, COUNT(DISTINCT geo) AS n_geos, COUNT(*) AS n_rows
FROM staging.stg_statcan__cpi
GROUP BY geo_level;

SELECT * FROM staging.stg_statcan__cpi
WHERE geo_code = 'CA' AND component = 'All-items'
ORDER BY month DESC
LIMIT 3;

Why is this a view and not a table? Staging is cheap to recompute from raw, and views never go stale. You'll materialise tables at the mart level.

18.6 Tests as Queries · A Test Runner (23 min)

Here's a simple and powerful convention, the one dbt uses: a test is a query that returns the rows that violate a rule. Zero rows = pass.

Create the folder sql/tests/ and add these files. The first line of each file declares its severity:

• error: the pipeline should stop
• warn: report it but continue

sql/tests/cpi_unique_keys.sql

-- severity: error
-- Each month × geo × component × base must appear once
SELECT month, geo, component, index_base, COUNT(*) AS n
FROM staging.stg_statcan__cpi
GROUP BY ALL
HAVING COUNT(*) > 1

sql/tests/cpi_positive_values.sql

-- severity: error
SELECT * FROM staging.stg_statcan__cpi WHERE cpi <= 0

sql/tests/cpi_no_gaps_core.sql

-- severity: error
-- National and provincial All-items series must have no missing months.
-- A series spanning N months from first to last must have exactly N rows.
SELECT geo, component,
       MIN(month) AS first_month,
       MAX(month) AS last_month,
       date_diff('month', MIN(month), MAX(month)) + 1 AS expected_rows,
       COUNT(*)                                       AS actual_rows
FROM staging.stg_statcan__cpi
WHERE geo_level IN ('national', 'province')
  AND component = 'All-items'
GROUP BY geo, component
HAVING date_diff('month', MIN(month), MAX(month)) + 1 <> COUNT(*)

That's a gap test without building a calendar spine: if there are no gaps, the count equals the number of months between first and last.

sql/tests/cpi_freshness.sql

-- severity: warn
-- The latest national CPI should be at most 3 months old
SELECT MAX(month) AS latest_month
FROM staging.stg_statcan__cpi
WHERE geo_code = 'CA' AND component = 'All-items'
HAVING MAX(month) < CAST(current_date - INTERVAL 3 MONTH AS DATE)

Don't end test files with a semicolon. The runner wraps each query inside another query.

The test runner: run_tests.py (repository root)

"""run_tests.py — run every sql/tests/*.sql against macro.duckdb.

A test passes when its query returns zero rows.
The first line '-- severity: error|warn' controls whether a failure stops the pipeline.
"""
import sys
from pathlib import Path

import duckdb

TEST_DIR = Path("sql/tests")


def severity_of(sql_text):
    first = sql_text.strip().splitlines()[0].lower()
    return "warn" if "severity: warn" in first else "error"


def main():
    con = duckdb.connect("macro.duckdb", read_only=True)
    errors = warnings = 0

    for path in sorted(TEST_DIR.glob("*.sql")):
        sql = path.read_text().strip().rstrip(";")
        severity = severity_of(sql)
        try:
            n = con.execute(f"SELECT COUNT(*) FROM ({sql}) AS t").fetchone()[0]
        except Exception as e:
            print(f"  ✗ ERROR  {path.stem}: query failed — {e}")
            errors += 1
            continue

        if n == 0:
            print(f"  ✓ pass   {path.stem}")
        elif severity == "warn":
            print(f"  ! warn   {path.stem}: {n} row(s)")
            warnings += 1
        else:
            print(f"  ✗ FAIL   {path.stem}: {n} row(s)")
            errors += 1

    con.close()
    print(f"\n{errors} error(s), {warnings} warning(s)")
    sys.exit(1 if errors else 0)


if __name__ == "__main__":
    main()

.quit                      # leave the CLI first (one writer at a time)
python run_tests.py

When a test fails, investigate with the same query in the CLI; it shows you exactly which rows break the rule.

Day 18 checklist

• I can explain what belongs in staging and what doesn't
• My raw layer keeps every source column, as text
• I profile a table before modelling it, and verify its real key
• I maintain reference data as versioned seeds, with comments in the database
• My staging view follows source → renamed → final and states its grain
• My tests are queries returning violations, run automatically with severities

git add .
git commit -m "Day 18: raw full tables, seeds, stg_statcan__cpi, SQL tests"
git push