The Ingestion Layer

How raw TPCH-shaped data actually lands: COPY INTO, Snowpipe, EL tools, loaded-at metadata, and the raw-to-staging contract.

Every pipeline in this series so far has read from SNOWFLAKE_SAMPLE_DATA.TPCH_SF1 — a share Snowflake maintains, keeps fresh, and hands you for free. That was a deliberate shortcut. It let us build staging models, marts, tests, and a Cosmos-rendered DAG without first solving the problem every real stack has to solve: the raw data doesn’t arrive by itself. Somebody has to land bytes in Snowflake before dbt has anything to stage.

And there’s a loose thread to tie off. Back in the scheduling chapter, the transform DAG was scheduled on an Asset — schedule=[Asset("snowflake://analytics_prod/raw/orders")] — so it fires when the data arrives instead of on a hopeful clock. But nothing in the series ever produced that asset. The consumer was written; the producer was hand-waved. This chapter writes the producer. By the end, a real ingest DAG lands TPCH-shaped orders into an ANALYTICS_PROD.RAW schema, declares that same asset as an outlet, and completing the load kicks off the whole dbt pipeline downstream.

From a borrowed share to your own raw

The first shift is conceptual. The sample data lives in a database Snowflake owns; your production data lives in a raw schema you own, inside your own database. Everything else in the stack — ANALYTICS_DEV and ANALYTICS_PROD, the roles, the warehouses — is yours too, and raw is the schema where the outside world’s data first touches your account. Nothing transforms it there. It’s a landing pad: columns in roughly the shape the source hands you, plus a little metadata about when it landed.

So the target for orders is a plain table in ANALYTICS_PROD.RAW, TPCH’s columns verbatim, with one column that isn’t in the source file at all:

create table ANALYTICS_PROD.RAW.orders (
    o_orderkey      number,
    o_custkey       number,
    o_orderstatus   varchar,
    o_totalprice    number(12,2),
    o_orderdate     date,
    o_orderpriority varchar,
    o_clerk         varchar,
    o_shippriority  number,
    o_comment       varchar,
    _loaded_at      timestamp_ntz default current_timestamp()
);

That _loaded_at column is the whole raw-to-staging contract in one line, and we’ll come back to it. First, the mechanics of getting rows into that table.

The file path: stage, file format, COPY INTO

The bread-and-butter way to load files into Snowflake is COPY INTO, and it needs two things defined first: a file format (how to parse the bytes) and a stage (where the files live).

A file format is a named, reusable description of the file’s shape. TPCH’s reference files are pipe-delimited with a trailing delimiter on each row — the classic .tbl format — so:

create file format ANALYTICS_PROD.RAW.tpch_csv
    type = csv
    field_delimiter = '|'
    skip_header = 0
    field_optionally_enclosed_by = '"'
    null_if = ('', 'NULL')
    empty_field_as_null = true
    trim_space = true
    compression = auto;

Defining it once means every COPY and every PIPE references format_name => tpch_csv instead of restating a dozen options and drifting apart over time.

A stage is Snowflake’s abstraction over a location full of files. There are two flavors. An internal stage is storage Snowflake manages for you — you push files to it with PUT from SnowSQL or a driver:

create stage ANALYTICS_PROD.RAW.tpch_internal
    file_format = ANALYTICS_PROD.RAW.tpch_csv;
# from SnowSQL, on the machine that has the files
put file:///data/tpch/orders/*.tbl @ANALYTICS_PROD.RAW.tpch_internal/orders/;

That’s fine for a bootstrap or an occasional manual load. Production ingestion almost always uses an external stage instead — a pointer at a cloud bucket (S3, GCS, ADLS) that some upstream process is already dropping files into. The catch is authentication: you don’t want AWS keys pasted into a CREATE STAGE. The clean answer is a storage integration, a Snowflake object that trusts an IAM role rather than holding a key:

create storage integration tpch_s3_int
    type = external_stage
    storage_provider = 's3'
    enabled = true
    storage_aws_role_arn = 'arn:aws:iam::123456789012:role/snowflake-tpch-loader'
    storage_allowed_locations = ('s3://acme-raw-landing/tpch/');

Then you run desc integration tpch_s3_int, read back the STORAGE_AWS_IAM_USER_ARN and STORAGE_AWS_EXTERNAL_ID Snowflake generated, and add those to the trust policy on the AWS side. The result is a bucket Snowflake can read without a single long-lived credential living in your account — the two sides trust each other by role, and either can rotate independently. The stage then names the integration instead of secrets:

create stage ANALYTICS_PROD.RAW.tpch_stage
    storage_integration = tpch_s3_int
    url = 's3://acme-raw-landing/tpch/'
    file_format = ANALYTICS_PROD.RAW.tpch_csv;

With the format and stage in place, the load itself is one statement — but the interesting part is that _loaded_at column, which isn’t in the file. You supply it with a transform in the COPY: instead of copying the stage straight into the table, you SELECT from the stage and add the timestamp as an extra expression.

copy into ANALYTICS_PROD.RAW.orders
    (o_orderkey, o_custkey, o_orderstatus, o_totalprice, o_orderdate,
     o_orderpriority, o_clerk, o_shippriority, o_comment, _loaded_at)
from (
    select
        $1, $2, $3, $4, $5, $6, $7, $8, $9,
        current_timestamp()
    from @ANALYTICS_PROD.RAW.tpch_stage/orders/
)
file_format = (format_name = ANALYTICS_PROD.RAW.tpch_csv)
pattern = '.*orders.*[.]tbl'
on_error = 'skip_file';

$1…$9 are the positional columns from each row; current_timestamp() stamps the moment Snowflake loaded it. The pattern is a regex that scopes the load to just the orders files under that stage prefix, so one shared bucket can feed every raw table without cross-contamination. If you also want lineage — which file did this row come from — Snowflake exposes it as metadata$filename, and you can carry it into a _source_file column the same way. _loaded_at answers “how fresh,” metadata$filename answers “from where”; together they make a raw table debuggable.

Idempotent by default (and how to break it on purpose)

Here’s the property that makes COPY INTO safe to re-run, which matters enormously once Airflow is retrying tasks: Snowflake remembers which files it already loaded. Every COPY records the files it consumed in the table’s load metadata, and it keeps that history for 64 days. Point the same COPY at the same stage again and it skips every file it’s already seen — a retried task re-loads nothing, a re-run after a partial failure only picks up the new files. You get idempotency for free, keyed on file identity, without writing a line of dedup logic.

Two escape hatches are worth knowing. FORCE = TRUE overrides the skip and re-loads everything — occasionally what you want after truncating a table, usually a foot-gun that produces duplicates, so reach for it deliberately. And before a big first load, VALIDATION_MODE = 'RETURN_ERRORS' runs the COPY as a dry run: it parses every file, returns the rows it would reject, and writes nothing. It’s the cheapest way to catch a wrong delimiter or an off-by-one column list before you’ve loaded a million bad rows.

The other lever is what happens when a row won’t parse. ON_ERROR decides:

  • ABORT_STATEMENT (the default) — one bad row fails the entire load. Correct when the file is supposed to be clean and a parse error means something is badly wrong upstream.
  • CONTINUE — skip the offending rows, load the rest. Maximizes throughput, quietly drops data.
  • SKIP_FILE — skip the whole file if it has any errors (or SKIP_FILE_5 / SKIP_FILE_5% for a threshold). The middle ground most batch loads want: a poisoned file doesn’t taint the load, and it doesn’t half-load either.

There’s no universally right answer — it’s a contract decision. A financial feed wants ABORT_STATEMENT and a human paged. A high-volume event firehose wants SKIP_FILE and a dashboard.

Watching the load

A load you can’t inspect is a load you don’t trust. Two views cover it. For recent activity, COPY_HISTORY in INFORMATION_SCHEMA is a table function scoped to one table:

select file_name, status, row_count, row_parsed, error_count, last_load_time
from table(information_schema.copy_history(
    table_name => 'ANALYTICS.RAW.ORDERS',
    start_time => dateadd('hour', -24, current_timestamp())
));

For a wider window there’s SNOWFLAKE.ACCOUNT_USAGE.COPY_HISTORY (account-level, up to 365 days, with the usual latency). And when a specific COPY rejected rows, validate() replays exactly what went wrong:

select * from table(validate(ANALYTICS_PROD.RAW.orders, job_id => '_last'));

That gives you the file, the row, the column, and the error for every rejection from the last load — the difference between “the load logged 400 errors” and “line 88,203 of orders_03.tbl has a date Snowflake can’t parse.”

Snowpipe: the same load, continuously

COPY INTO is a batch verb — you run it, it loads what’s there, it stops. When files arrive continuously and you want them loaded within a minute or two of landing rather than on the next scheduled run, that’s Snowpipe. A pipe is a named, saved COPY statement that Snowflake runs for you when new files show up:

create pipe ANALYTICS_PROD.RAW.orders_pipe
    auto_ingest = true
as
copy into ANALYTICS_PROD.RAW.orders
    (o_orderkey, o_custkey, o_orderstatus, o_totalprice, o_orderdate,
     o_orderpriority, o_clerk, o_shippriority, o_comment, _loaded_at)
from (
    select $1, $2, $3, $4, $5, $6, $7, $8, $9, current_timestamp()
    from @ANALYTICS_PROD.RAW.tpch_stage/orders/
)
file_format = (format_name = ANALYTICS_PROD.RAW.tpch_csv);

AUTO_INGEST = TRUE wires the pipe to cloud event notifications: you run desc pipe orders_pipe, take the notification_channel SQS ARN it returns, and configure your S3 bucket to publish s3:ObjectCreated events to it. Now every new object in the stage triggers the pipe automatically — no scheduler involved. (There’s also a REST path via insertFiles if you’d rather push notifications yourself.) Snowpipe is serverless: it runs on Snowflake-managed compute, billed per-file rather than by a warehouse you keep awake, which is exactly the right cost model for a trickle of files arriving at odd hours.

The trade-off is orchestration visibility. A pipe fires outside Airflow, so it doesn’t update an Airflow asset — which is a problem for us, because the entire point of this chapter is producing the asset that triggers the transform DAG. Two ways to reconcile that: keep the load inside Airflow as a COPY task (what we do below, and the cleanest loop), or let Snowpipe do the loading and add a small Airflow task that watches a stream on raw.orders and emits the asset when new rows appear. For a daily-batch TPCH pipeline, batch COPY in Airflow is the honest fit. Save Snowpipe for the genuinely event-shaped feeds.

Where a managed EL tool fits

Everything above is you hand-rolling extraction and load. The industry mostly doesn’t — it buys it. Fivetran, Airbyte, and their kin are managed EL tools: you point a connector at a source (a Postgres database, Stripe, Salesforce, a SaaS API), and it handles extraction, schema detection, incremental syncs, and schema drift, landing the result in your warehouse. What they land is still a raw schema of source-shaped tables — the contract downstream is identical. dbt neither knows nor cares whether raw.orders arrived via Fivetran or via a COPY you wrote; it sees a table with columns and a load timestamp.

So when do you still hand-roll COPY INTO? When you already have files in a bucket and no connector is involved — a partner drops a nightly extract in S3, and there’s nothing to “connect” to. When the source is bespoke enough that no connector exists. When the volume is large enough that a per-row-priced connector gets expensive and a COPY off a bucket is nearly free. And for one-time bulk backfills, where standing up a connector to load history once is more work than a COPY with a pattern. The rule of thumb: buy the connectors for the long tail of SaaS sources where drift and API churn would eat your week; hand-roll the file loads where you already own the files and want the control. Either way, raw looks the same, and so does the producer’s job — land the data, then tell the rest of the stack it arrived.

Closing the loop: the producer DAG

Now the payoff. The transform DAG waits on Asset("snowflake://analytics_prod/raw/orders"). To make it fire, some task somewhere has to declare that asset in its outlets and succeed. That task is a COPY INTO running on the same snowflake_default connection every Snowflake step in this series has used:

from datetime import datetime
from airflow.sdk import dag, Asset
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

RAW_ORDERS = Asset("snowflake://analytics_prod/raw/orders")

COPY_ORDERS = """
copy into ANALYTICS_PROD.RAW.orders
    (o_orderkey, o_custkey, o_orderstatus, o_totalprice, o_orderdate,
     o_orderpriority, o_clerk, o_shippriority, o_comment, _loaded_at)
from (
    select $1, $2, $3, $4, $5, $6, $7, $8, $9, current_timestamp()
    from @ANALYTICS_PROD.RAW.tpch_stage/orders/
)
file_format = (format_name = ANALYTICS_PROD.RAW.tpch_csv)
pattern = '.*orders.*[.]tbl'
on_error = 'skip_file';
"""

@dag(schedule="@hourly", start_date=datetime(2026, 6, 1), catchup=False)
def ingest_raw_orders():
    SQLExecuteQueryOperator(
        task_id="copy_into_orders",
        conn_id="snowflake_default",
        sql=COPY_ORDERS,
        outlets=[RAW_ORDERS],
    )

ingest_raw_orders()

That outlets=[RAW_ORDERS] is the entire mechanism. When copy_into_orders succeeds, Airflow materializes the asset — records a new update — and any DAG scheduled on it wakes up. The transform DAG from chapter 6 is scheduled on exactly this asset, so the sequence is now complete end to end: files land in the stage, this hourly DAG COPYs them into raw.orders, the asset updates, and the Cosmos-rendered dbt pipeline runs because the orders arrived, not because a clock struck. If ingest is late, the transform waits. If ingest lands nothing new (the COPY skips already-loaded files but the task still succeeds), the transform still fires but the incremental models process an empty delta — cheap and harmless.

A word on which warehouse runs that COPY. The snowflake_default connection defaults to the transforming warehouse, and a load will run there perfectly well — but loading and transforming are different workloads with different shapes, and the cost chapter later in this series makes the case for splitting them. A loading warehouse sized XSMALL is usually plenty for COPY, which is I/O-bound rather than compute-bound; you don’t need a big warehouse to parse files. Keeping loads off the transform warehouse also stops a heavy dbt build and a bulk load from queuing behind each other on the same compute. You can steer a single task’s warehouse without touching the connection by prefixing the SQL — use warehouse loading; copy into ... — or, cleaner, by giving ingest its own connection whose default warehouse is loading. Same key pair, same account, different compute lane.

One detail is doing all the work and it’s easy to miss: the URI is an opaque string, and it must match byte-for-byte. snowflake://analytics_prod/raw/orders here has to be character-identical to the string chapter 6 passed to Asset(...). Airflow never parses that URI, never connects to Snowflake through it, never validates it points anywhere real — it’s just a name two DAGs agree on. Producer and consumer are wired together entirely by string equality. Fat-finger raw/order in one place and the two DAGs silently never talk, with no error to tell you why. In a real project you’d define the Asset once in a shared module and import it into both DAGs, precisely so the string can’t drift.

You rarely load just one table. Real ingest fans the pattern out — a COPY task per raw table, each with its own outlet — and the transform DAG decides how many of those it waits on. Airflow 3.x lets you combine assets in a schedule: schedule=[orders, lineitem, customer] is an implicit all-of (the transform waits for every one to update), while schedule=(orders & lineitem) | reload expresses richer logic — build when the two fact-ish tables both refresh, or whenever a manual reload asset is poked. For TPCH the natural shape is “wait for orders and lineitem, treat the slow-moving dimension tables as best-effort,” so the fact loads gate the run and a stale nation doesn’t block anything. The producer side stays dead simple regardless: each COPY task declares its one outlet, and the scheduling logic lives entirely in the consumer’s schedule=. You can also emit richer signal than “it happened” by running the COPY through a TaskFlow task, reading back the rows the load reported, and attaching that as asset metadata:

from airflow.sdk import task, Metadata
from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook

@task(outlets=[RAW_ORDERS])
def copy_into_orders():
    hook = SnowflakeHook(snowflake_conn_id="snowflake_default")
    # COPY returns one row per file: name, status, rows_parsed, rows_loaded, ...
    results = hook.get_records(COPY_ORDERS)
    loaded = sum(row[3] for row in results)   # rows_loaded column
    files = len(results)
    yield Metadata(RAW_ORDERS, {"rows_loaded": loaded, "files_loaded": files})

Now the asset event carries how many rows landed, visible in the Airflow UI next to the update — the difference between “orders refreshed” and “orders refreshed: 12,480 rows across 3 files.”

The raw-to-staging contract

_loaded_at isn’t decoration. It’s the term in the contract that lets the transform layer reason about freshness without knowing anything about how the data got here. dbt formalizes it in the source definition. Remember that in chapter 5 staging read from the SNOWFLAKE_SAMPLE_DATA share; in production it reads from your own raw schema, and the source declaration is where you also state your freshness expectations:

# models/staging/_tpch__sources.yml
sources:
  - name: raw
    database: analytics
    schema: raw
    loaded_at_field: _loaded_at
    freshness:
      warn_after: {count: 12, period: hour}
      error_after: {count: 24, period: hour}
    tables:
      - name: orders
      - name: lineitem
      - name: customer
      - name: nation
      - name: region

loaded_at_field: _loaded_at tells dbt which column marks arrival time. The freshness block turns that into an SLA: dbt source freshness runs select max(_loaded_at) from ANALYTICS_PROD.RAW.orders, compares it to now, and warns if the newest row is over 12 hours old, errors if it’s over 24. Warn logs and keeps going; error fails the check — so you can gate the transform DAG on it and refuse to build marts on top of stale raw data rather than publishing yesterday’s numbers as though they were today’s. On a big table you’d add a filter: to the freshness config so that max() scan stays cheap, and per-table overrides let a fast-moving orders demand tighter freshness than a nation table that changes once a year.

The staging model then reads raw through source(), exactly as it read the sample share — the only change is the source name:

-- models/staging/stg_orders.sql
select
    o_orderkey    as order_id,
    o_custkey     as customer_id,
    o_orderstatus as status,
    o_totalprice  as total_price,
    o_orderdate   as order_date
from {{ source('raw', 'orders') }}

Notice staging drops _loaded_at — it’s raw-layer metadata, not a business column, and nothing downstream should join on it. It exists to answer freshness and to debug “when did this row arrive,” and its job ends at the source boundary.

A contract is more than a timestamp, though, and the honest version writes down the rest. Is raw.orders append-only, or does each load fully replace the table? That determines whether staging can trust that a missing key means a deleted order or just a not-yet-loaded one — TPCH batch files are append-and-update, so deletes aren’t represented at all, and any “soft delete” logic has to come from a status column, not from absence. How does schema drift show up? With hand-rolled COPY and a fixed column list, a new source column is silently ignored until you widen the list — safe but invisible; with a managed EL tool, new columns usually appear automatically, which is convenient until an unexpected one breaks a downstream assumption. None of these have a universally right answer. What matters is that the contract states one, so the person debugging a stale mart at 2 a.m. knows whether they’re looking at a load that didn’t run, a delete that isn’t represented, or a column that drifted.

Final thoughts

The sample-data share let us pretend raw data is free and always fresh. It isn’t. Something lands it, on a cadence, with a shape and a freshness and an error policy — and every one of those is a decision you make, not a default you inherit. The load itself is almost anticlimactic: a file format, a stage, one COPY INTO, idempotent because Snowflake remembers the files it’s seen. The engineering is in the seams around it. _loaded_at and a source freshness SLA turn “is this data current?” from a guess into a query. The outlets=[Asset(...)] on the load task turns “did ingest finish?” from a Slack message into a trigger. That single outlet is what finally makes the whole stack one connected graph instead of two DAGs that happen to touch the same table and hope their timers line up — the producer announces the data arrived, and the transform runs because it did. The asset was always there waiting to be produced. Now something produces it.

Next: Environments and RBAC Across the Stack

Comments