ETL Pipeline Design skill for Codex

Done well means: any run can be repeated for the same window with the same result, raw input is kept so history can be rebuilt, late data has a defined window and path, schema changes fail loudly or evolve deliberately, and each run records what it read and wrote.

Before designing

1. Read AGENTS.md and any existing pipeline code, orchestrator config and warehouse DDL.
2. Write down, with the user, the answers to these questions; the design follows from them:

   - [ ] Sources: system, access method, volume per day, can it be re-read later?
   - [ ] Change signal: append-only, updated_at column, soft deletes, hard deletes, CDC log?
   - [ ] Freshness SLA for consumers (hourly, daily by 07:00)?
   - [ ] How late can records arrive? (minutes, days, months for corrections)
   - [ ] Who consumes the output and what breaks if a run is wrong?
   - [ ] Retention and PII constraints on raw data

ETL vs ELT

Situation	Choose
Cloud warehouse or lakehouse with cheap compute (Snowflake, BigQuery, Databricks, DuckDB)	ELT: land raw, transform with SQL/dbt in the warehouse
PII must be removed or masked before it lands anywhere central	ETL step for masking, then ELT
Heavy parsing (binary formats, documents, API pagination)	Extract-and-normalise in code, then ELT
Target is an operational system (CRM, search index)	Reverse ETL from modeled tables

Default to ELT with a raw layer. It makes reprocessing a re-run of SQL instead of re-extraction.

Extraction strategy

Source behaviour	Strategy	Notes
Small table (< a few million rows)	Full snapshot each run	Simplest; detects hard deletes by comparison
Reliable updated_at, soft deletes	Incremental watermark with overlap	updated_at >= last_watermark - overlap
Hard deletes or no reliable timestamp	CDC from the database log	Hand off to $cdc-debezium
Append-only events	Partition by event date, read new partitions or offsets	Dedupe on event ID
Third-party API	Paginate by time window, store raw JSON responses	Respect rate limits; persist cursors

Watermark rules:

• Store the watermark after the load commits, never before.
• Use the source’s clock (max updated_at seen), not the pipeline’s now().
• Re-read an overlap (for example 1 hour or 1 day) and dedupe on the key; transactions that commit late carry older updated_at values.

Layers

Layer	Content	Write mode
Raw / landing	Exact source payload plus _loaded_at, _source_file, _run_id	Append-only, immutable, partitioned by load date
Staging	Typed, renamed, deduplicated per key	Rebuilt or overwritten per window
Modeled / marts	Business grain facts and dimensions	Partition overwrite or MERGE

Keep raw for at least the longest reprocessing horizon you expect; that is what makes logic fixes cheap.

Idempotent loads

Every load is keyed on a window or a key, never “whatever arrived since last time” appended blindly.

-- partition overwrite: the window is the unit of work
begin;
delete from marts.daily_orders where order_date = :run_date;
insert into marts.daily_orders
select order_date, customer_id, count(*), sum(amount)
from staging.orders where order_date = :run_date
group by 1, 2;
commit;

-- upsert for updatable entities
merge into marts.customers t
using (select * from staging.customers
       qualify row_number() over (partition by customer_id order by updated_at desc) = 1) s
on t.customer_id = s.customer_id
when matched and s.updated_at > t.updated_at then update set name = s.name, updated_at = s.updated_at
when not matched then insert (customer_id, name, updated_at) values (s.customer_id, s.name, s.updated_at);

Deduplicate the source side of a MERGE; duplicate source keys either error or double-apply.

Late-arriving data

1. Define a lateness window from data (for example 99.9% of events arrive within 3 days).
2. Each run reprocesses the current window plus the lateness window (run_date - 3 .. run_date) with partition overwrite. Cost is bounded and results converge.
3. Records later than the window: land in raw as usual and get picked up by a scheduled weekly or monthly reconciliation run, or flagged for a manual backfill.
4. Late dimension rows: insert a placeholder “unknown” member keyed on the natural key and update it when the real row arrives, instead of dropping facts.

Reprocessing and backfills

When logic changes or a bug is found:

1. Fix the code and add a test that reproduces the bug.
2. Determine the affected window from the bug’s introduction date.
3. Rebuild from raw into a shadow table or schema for that window; compare against current output (row counts, key totals, sample diffs).
4. Present the diff and the cost estimate to the user and get explicit approval before swapping into production tables or running a large backfill.
5. Swap (rename, partition exchange, or overwrite per partition), then notify consumers.

Run backfills with bounded concurrency so production runs keep their resources.

Schema drift

Change upstream	Policy
New nullable column	Land in raw automatically (semi-structured or schema evolution); add downstream deliberately
Column removed or renamed	Fail the staging step loudly; never fill silently with nulls
Type widened (int to bigint)	Accept and widen the target
Type changed incompatibly	Fail; route to a human

Store raw payloads in a format that tolerates drift (JSON/VARIANT column, Parquet with schema evolution) so the raw layer never rejects data.

Run metadata

Record per run: run_id, window start and end, source watermark before and after, rows read, rows written, rows rejected, duration, code version (git SHA), status. Store it in a table. It answers “what changed and when” during incidents and powers volume checks.

Gotchas

• now() inside transformations makes reruns produce different results; pass the window in.
• Timezones: store UTC, define the business-day boundary explicitly, and say which one a daily partition uses.
• Retries of a step that appends create duplicates; retries of an overwrite do not.
• Deletes in the source are invisible to updated_at extraction; decide how they propagate.
• Running two instances of the same pipeline for the same window races; use orchestrator concurrency limits or a lock.

Design doc template

**Pipeline:** billing_invoices -> warehouse (daily by 06:00 UTC)
**Extraction:** incremental on updated_at, 2 h overlap, watermark stored after commit; hard deletes via nightly key diff
**Layers:** raw.billing_invoices (JSON, append, 400-day retention) -> stg (dedup by invoice_id) -> fct_invoices (MERGE)
**Late data:** reprocess last 3 days each run; monthly reconciliation for older corrections
**Idempotency:** rerun of any date yields identical counts (tested 2025-09-01 twice)
**Schema drift:** new columns land in raw; removed columns fail stg with alert
**Run metadata:** ops.pipeline_runs; volume check +/-30% vs 7-day median
**Backfill plan:** shadow rebuild 2025-06-01..2025-09-30, diff reviewed, swap after approval