Building the TPCH Pipeline

Everything until now was scaffolding — a profile, a connection, a bridge. This is the post where you build the thing the scaffolding was holding up: a dbt project that turns the raw TPCH tables into marts a business would actually query, running on Snowflake, orchestrated end to end by Airflow. No more diagrams. Models with SQL in them.

Three layers, not two

The instinct is to build two layers — staging on the raw data, marts on staging — and for a project this small you could get away with it. You shouldn’t, and the reason is a join you’re about to write twice. Both marts in this pipeline need to walk lineitem → orders → customer → nation → region, and if that walk lives inside each mart, you now maintain it in two places and re-run it in two places. So the project has three layers, and the middle one exists precisely to make sure nobody walks that join more than once.

• Staging sits directly on the raw sample data — one model per source table — and does nothing but clean, rename, cast, and key. Nothing downstream ever touches SNOWFLAKE_SAMPLE_DATA again; it goes through staging, which means the day the source schema changes you fix it in one model instead of forty.
• Intermediate (int_) is where the shared join and the risky arithmetic live once. It has no consumers outside the project — no BI tool queries it — so it doesn’t even need to be a table.
• Marts are the tables a human or a dashboard reads. They’re thin: by the time you reach a mart, the joining and the aggregating are already done, and the mart is a final projection with a business name on it.

The raw tables are declared once, as a source, and this is also where you wire up the first thing that isn’t a model at all — freshness:

# models/staging/_tpch__sources.yml
sources:
  - name: tpch
    database: SNOWFLAKE_SAMPLE_DATA
    schema: TPCH_SF1
    tables:
      - name: orders
        loaded_at_field: "o_orderdate::timestamp"
        freshness:
          warn_after: { count: 48, period: hour }
          error_after: { count: 168, period: hour }
      - name: lineitem
      - name: customer
      - name: nation
      - name: region

dbt source freshness compiles that block into a query that asks Snowflake for max(o_orderdate) and measures how long ago that was. A blunt honesty is owed here: TPCH_SF1 is a static sample whose orders are years old, so this freshness check would fire on the first run. That’s not a bug in the config — it’s the config working, telling you the “source” hasn’t loaded recently. On the real ingested landing tables you’ll build two posts from now, the same block is the tripwire that stops the whole pipeline from transforming yesterday’s data as if it were today’s. It’s worth wiring in now so the shape is already there when the data underneath it starts moving.

Staging: cast once, rename once, never again

Every staging model has the same skeleton, and the sameness is the point — anyone who opens stg_* knows exactly what they’ll find. Pull the source in one CTE, do the boring work in a second, and select the result:

-- models/staging/stg_orders.sql
with source as (
    select * from {{ source('tpch', 'orders') }}
),

renamed as (
    select
        o_orderkey              as order_id,
        o_custkey               as customer_id,
        o_orderstatus           as status,
        o_totalprice::number(12, 2) as total_price,
        o_orderdate::date       as order_date
    from source
)

select * from renamed

Two things earn their place in that second CTE. The rename turns TPCH’s cryptic o_-prefixed columns into names a person can read — every model downstream says order_date, never o_orderdate. The cast is where you refuse to inherit whatever type the source happened to land in. o_totalprice is money, so it becomes number(12, 2) and stops there; o_orderdate is a calendar date, not a timestamp with a phantom midnight on it, so it becomes date. TPCH is disciplined about types and you might get away with skipping this, but the habit is what saves you when a source column arrives as a string that looks like a number. Cast it once, at the edge, and everything downstream can trust it.

lineitem is where casting stops being decorative. Revenue in this benchmark comes out of l_extendedprice, a decimal — and if you let Snowflake infer its way through l_extendedprice * (1 - l_discount) without pinning precision, rounding drifts on you at scale. Pin it. Line items also expose the other thing staging owes you: a key. TPCH’s line-item grain is (l_orderkey, l_linenumber) — there is no single natural primary key — so you mint a surrogate from the pair with dbt_utils:

-- models/staging/stg_lineitem.sql
with source as (
    select * from {{ source('tpch', 'lineitem') }}
),

renamed as (
    select
        {{ dbt_utils.generate_surrogate_key(['l_orderkey', 'l_linenumber']) }} as line_item_id,
        l_orderkey                   as order_id,
        l_linenumber                 as line_number,
        l_partkey                    as part_id,
        l_suppkey                    as supplier_id,
        l_quantity::number(12, 2)    as quantity,
        l_extendedprice::number(12, 2) as extended_price,
        l_discount::number(12, 2)    as discount,
        l_shipdate::date             as ship_date
    from source
)

select * from renamed

generate_surrogate_key hashes the concatenation of l_orderkey and l_linenumber into one stable string. Now a line item has a single-column identity you can test unique and not_null on, and you can point a relationships test at it later without a compound-key song and dance. stg_customer follows the same skeleton — c_custkey as customer_id, c_nationkey as nation_id, c_acctbal::number(12,2) as account_balance — and stg_nation_region is the one staging model that does more than one table’s worth of work, folding the two-hop nation → region lookup into a single flat model so nothing downstream has to remember that geography is a join:

-- models/staging/stg_nation_region.sql
with nation as (
    select * from {{ source('tpch', 'nation') }}
),

region as (
    select * from {{ source('tpch', 'region') }}
)

select
    nation.n_nationkey as nation_id,
    nation.n_name      as nation,
    region.r_regionkey as region_id,
    region.r_name      as region
from nation
join region on nation.n_regionkey = region.r_regionkey

The intermediate layer nobody skips twice

Here’s the join both marts want, built exactly once. And building it once forces you to confront the trap that would otherwise be hiding inside orders_enriched: grain.

orders_enriched is supposed to have one row per order. revenue_by_region sums money that only exists at the line-item level. Those are two different grains, and line item is the finer one — a single order fans out to many line items. If you write orders_enriched as a naive join of orders to line items to get revenue onto it, every order silently multiplies into as many rows as it has line items, total_price gets counted once per line, and every downstream sum is inflated by a factor nobody can see. That’s the classic fan-out bug, and it doesn’t error — it just quietly returns numbers that are too big.

The fix is to aggregate line items to order grain first, in the intermediate layer, before anything joins them to orders. One int_ model does the collapse and the shared join in the same breath:

-- models/intermediate/int_orders_enriched.sql
with order_revenue as (
    select
        order_id,
        sum(extended_price * (1 - discount)) as order_revenue,
        count(*)                             as line_item_count
    from {{ ref('stg_lineitem') }}
    group by order_id
),

orders_with_geography as (
    select
        o.order_id,
        o.customer_id,
        o.status,
        o.order_date,
        c.customer_name,
        nr.nation,
        nr.region
    from {{ ref('stg_orders') }}      o
    join {{ ref('stg_customer') }}    c  on o.customer_id = c.customer_id
    join {{ ref('stg_nation_region') }} nr on c.nation_id  = nr.nation_id
)

select
    og.order_id,
    og.customer_id,
    og.customer_name,
    og.status,
    og.order_date,
    og.nation,
    og.region,
    rev.order_revenue,
    rev.line_item_count
from orders_with_geography og
join order_revenue rev on og.order_id = rev.order_id

Read the grain of each CTE. order_revenue is one row per order — the group by order_id collapses the fan-out before it can touch anything else. orders_with_geography is one row per order too, because orders-to-customer-to-region is all many-to-one on the way down. Join two order-grained sets on order_id and you get one clean row per order, carrying both the region label and the summed revenue. The fan-out never happens, because the line items were already folded up when they arrived.

To feel how much this matters, put a number on it. TPCH orders carry between one and seven line items, averaging about four. Join orders straight to line items and orders_enriched isn’t a 1.5-million-row table anymore — it’s a six-million-row one, and every total_price in it appears four-ish times. Any analyst who then does sum(total_price) reports a revenue figure roughly quadruple the truth, and there is nothing in the output that looks wrong: the numbers are plausible, the rows are populated, the dashboard renders. That’s the whole danger of a grain bug — it doesn’t throw, it lies. Collapsing line items to order grain in the intermediate layer is the one move that makes the lie impossible, and doing it here means both marts inherit the fix for free.

And this is the model both marts stand on. The lineitem → orders → customer → nation → region walk lives here, and only here. Neither mart re-derives it — one of them slices it by region, the other just publishes it.

Two marts on one shared spine

With the join and the aggregation already done, the marts are almost embarrassingly short. revenue_by_region is the model the whole benchmark exists to make you write, and now it’s a group by over an intermediate that already knows the region:

-- models/marts/revenue_by_region.sql
select
    region,
    sum(order_revenue)     as revenue,
    sum(line_item_count)   as line_items,
    count(*)               as orders
from {{ ref('int_orders_enriched') }}
group by region
order by revenue desc

No joins in the mart at all — the joins are upstream. The mart is the business question stated as arithmetic: revenue by region, plus the order and line-item counts a stakeholder always asks for two seconds later.

The second mart is orders_enriched, and it’s the analyst-facing wide table — one row per order with the customer, the geography, and the totals already attached, so nobody slicing orders ever has to relearn the join:

-- models/marts/orders_enriched.sql
select
    order_id,
    customer_id,
    customer_name,
    status,
    order_date,
    nation,
    region,
    order_revenue,
    line_item_count
from {{ ref('int_orders_enriched') }}

That’s it — a projection over the shared intermediate. It looks trivial, and it is supposed to look trivial: all the danger was in the grain, and the grain got handled one layer down where both marts benefit from the fix. This is also the exact model the next post turns incremental — when int_orders_enriched carries three years of history, rebuilding this table from scratch every night is the waste we go after. For now it’s a full table, and it’s correct.

Because every reference here is a ref(), dbt infers the whole dependency order — four staging models, then the intermediate, then the two marts — and Cosmos turns that inferred order into the Airflow graph. You never wrote an edge by hand.

Materialization is a cost decision

On a laptop, how a model is stored barely matters. On Snowflake, it’s a line item on your bill, and the three layers each want a different answer. Set it once per folder in dbt_project.yml rather than sprinkling config() blocks across forty models:

# dbt_project.yml
models:
  tpch:
    staging:
      +materialized: view
      +schema: staging
      +tags: ["staging"]
    intermediate:
      +materialized: ephemeral
      +tags: ["intermediate"]
    marts:
      +materialized: table
      +schema: marts
      +tags: ["marts"]

Walk the reasoning, because it’s the same reasoning you’ll apply to every dbt project on Snowflake you ever build:

• Staging is a view. A view stores no data and costs nothing to build — it’s just a saved query. The only thing that reads a staging model is the next model in the same dbt build, so there’s no point paying to materialize rows that get consumed seconds later. The compute happens once, at build time, folded into the model that selects from it.
• Intermediate is ephemeral. An ephemeral model isn’t even a database object — dbt inlines it as a CTE directly into whatever refs it. int_orders_enriched never becomes a table or a view; its SQL is pasted into revenue_by_region and orders_enriched at compile time. You get the reuse and the single-source-of-truth in your codebase without paying to persist a middle layer nobody queries directly. (The one cost: ephemeral models don’t show up as their own object, so you can’t test them with a stored relationships test the same way — which is part of why the keys and casts live in staging, where they can be tested, not here.)
• Marts are tables. These are the objects a dashboard hits, possibly hundreds of times a day. Materializing them means you pay the compute once per build and every read after that is a cheap scan of stored rows. The alternative — a mart as a view — would re-run the entire join-and-aggregate on every single dashboard load, which on Snowflake means spinning the warehouse for every curious analyst. Pay once, read cheap.

That’s the whole Snowflake cost story in one config block: push compute to build time where you control it, and never make a human wait on a warehouse for a number you could have stored. The +tags earn their keep in the next post — they let Airflow select tag:marts for a targeted rebuild without naming models.

ephemeral is the right default for int_orders_enriched, but it’s a default worth knowing when to abandon. Because an ephemeral model is inlined into every mart that refs it, its SQL runs once per consumer — with two marts, the join-and-aggregate executes twice per build. At this scale that’s nothing. If a fourth and fifth mart start leaning on the same intermediate, or if the aggregation itself grows expensive, you’re paying to recompute the identical result several times over, and the answer flips: promote the intermediate to a table (or a view), pay for the join once, and let the marts read the stored result. That’s the same push-compute-to-build-time logic applied one layer up — the only reason it isn’t the call today is that two cheap inlinings are cheaper than one persisted object you’d have to store and test.

Tests that catch the mistakes this pipeline makes

Tests in dbt aren’t a coverage metric; they’re assertions about the specific ways this data can be wrong. Cosmos renders each one as its own task, so a failed test is a discrete red square in the grid, downstream of the model it guards. The interesting tests are the ones aimed at a real hazard:

# models/staging/_stg__models.yml
models:
  - name: stg_orders
    description: "One row per order, cleaned and typed from the raw TPCH orders table."
    columns:
      - name: order_id
        description: "Primary key — TPCH o_orderkey."
        tests: [unique, not_null]
      - name: customer_id
        description: "Foreign key into stg_customer."
        tests:
          - not_null
          - relationships:
              to: ref('stg_customer')
              field: customer_id
      - name: status
        tests:
          - accepted_values:
              values: ["O", "F", "P"]

  - name: stg_lineitem
    columns:
      - name: line_item_id
        tests: [unique, not_null]
      - name: discount
        tests:
          - dbt_expectations.expect_column_values_to_be_between:
              min_value: 0
              max_value: 1

Each test is pointed at a failure that would otherwise be invisible:

• relationships on customer_id is the one this pipeline is genuinely prone to. Every mart join assumes an order’s customer exists; this test compiles to a query returning any customer_id with no match in stg_customer. If it returns a row, the marts that depend on that key never build on top of a silent orphan.
• accepted_values on status catches a source that starts emitting a fourth status code you didn’t design for. TPCH orders are O, F, or P; the day a C shows up, you want a red square, not a mystery bucket in a dashboard.
• dbt_expectations.expect_column_values_to_be_between on discount guards the arithmetic directly. Revenue is extended_price * (1 - discount); a discount above 1 would flip the sign of every affected line and quietly turn revenue negative. Asserting 0 ≤ discount ≤ 1 is asserting that the formula stays sane.

Both dbt_utils and dbt_expectations come from packages.yml and a dbt deps — the same install that gave you generate_surrogate_key:

# packages.yml
packages:
  - package: dbt-labs/dbt_utils
    version: [">=1.3.0", "<2.0.0"]
  - package: calogica/dbt_expectations
    version: [">=0.10.0", "<0.11.0"]

The marts get their own guardrails, and they’re different in kind — a mart test asserts the result is shaped right, not that an input is clean:

# models/marts/_marts__models.yml
models:
  - name: revenue_by_region
    description: "Total revenue and order counts rolled up to TPCH's five regions."
    columns:
      - name: region
        tests: [unique, not_null]
      - name: revenue
        tests:
          - dbt_expectations.expect_column_values_to_be_between:
              min_value: 0
  - name: orders_enriched
    columns:
      - name: order_id
        tests: [unique, not_null]

unique on revenue_by_region.region is the fan-out canary — if the grain ever breaks and a region shows up twice, this test fails loudly instead of double-counting into a chart. unique on orders_enriched.order_id asserts the thing the whole intermediate layer was built to guarantee: one row per order. And revenue ≥ 0 is the downstream echo of the discount check — if bad discounts ever slip through, negative revenue trips a mart test even if the staging test somehow didn’t.

Freshness rounds it out. dbt source freshness is a separate command from dbt build, and Cosmos can render it as its own task at the head of the graph — a gate that refuses to transform stale source data at all. On static TPCH it’s a teaching prop; on real ingestion it’s the difference between a pipeline that notices the upstream feed died and one that cheerfully republishes yesterday.

Bracketing the models with Snowflake steps

Rendered as a DbtDag, the dbt project is the whole DAG. But a real run usually wants a step or two that isn’t a dbt model — resize the warehouse before a heavy build, size it back after. That’s why you reach for DbtTaskGroup inside your own @dag, with plain Snowflake operators on either side:

from datetime import datetime
from airflow.sdk import dag
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
from cosmos import DbtTaskGroup, ProjectConfig, ExecutionConfig, RenderConfig
from tpch_profile import profile_config  # the ProfileConfig from post 4

@dag(schedule="@daily", start_date=datetime(2026, 6, 1), catchup=False)
def tpch_pipeline():
    resize_up = SQLExecuteQueryOperator(
        task_id="warehouse_up",
        conn_id="snowflake_default",
        sql="alter warehouse transforming_wh set warehouse_size = 'MEDIUM'",
    )

    transform = DbtTaskGroup(
        group_id="dbt_tpch",
        project_config=ProjectConfig("/opt/airflow/dbt/tpch"),
        profile_config=profile_config,
        execution_config=ExecutionConfig(dbt_executable_path="/opt/airflow/dbt-venv/bin/dbt"),
        render_config=RenderConfig(select=["staging", "intermediate", "marts"]),
    )

    resize_down = SQLExecuteQueryOperator(
        task_id="warehouse_down",
        conn_id="snowflake_default",
        sql="alter warehouse transforming_wh set warehouse_size = 'XSMALL'",
    )

    resize_up >> transform >> resize_down

tpch_pipeline()

Both operators and the whole task group ride the same snowflake_default connection — the one that’s held your key-pair credentials since post 3. The >> chain reads as the run actually behaves: size the warehouse up, let every staging, intermediate, and mart model run as its own retryable task on that warehouse, then size it back down so an idle warehouse doesn’t quietly bill you. In the grid it’s one graph — a Snowflake step, a fan-out of dbt models with their tests hanging off them, a Snowflake step — top to bottom.

Final thoughts

Look at what each tool contributed and none of it overlaps. dbt owns the SQL, the layering, and the dependency order. Cosmos owns the translation from that order into tasks. Airflow owns the schedule, the retries, and the two warehouse steps dbt has no business knowing about. Snowflake owns every byte of compute. You didn’t write a single line that made two of those tools care about each other’s internals — the ref() graph and the one connection did all the coordinating.

The pipeline is also correct in the ways that matter and cheap in the ways you control: the fan-out is caught one layer before it can inflate a number, the join is written once and reused twice, and the storage cost of each layer is a deliberate choice rather than an accident. What it isn’t yet is scaled. This all ran on TPCH_SF1 — about six million line items. Snowflake ships the same schema at TPCH_SF10, TPCH_SF100, and TPCH_SF1000, and at scale factor 1000 that line-item table is roughly six billion rows. Point this exact project at TPCH_SF1000 and the nightly full rebuild of orders_enriched stops being free and starts being a decision you’d get asked about in a cost review. That’s the next post: making the heavy models incremental so they process only what changed — a change that touches the dbt layer and leaves everything else in this DAG exactly as it is.

Next: Incremental Models and Scheduling on Snowflake