CI/CD for Data Pipelines

DEV — individual developer sandbox
  Data: subset of production (last 7 days). Schema: dev_{developer_name}
  Isolation: complete — dev changes cannot affect staging or prod
  Lifespan: created on branch checkout, deleted after merge

STAGING / CI — automated testing environment
  Data: clone of production (Zero-Copy Clone). Schema: ci_{PR_number}
  Isolation: each PR gets its own schema
  Lifespan: created on PR open, deleted after PR merge

PRODUCTION — serves real analysts and BI tools
  Data: full production data, updated by live pipelines
  Access: pipeline service accounts write; analysts read-only
  Lifespan: permanent

freshcart:
  target: dev

  outputs:
    dev:
      type: snowflake
      account: freshcart.snowflake.com
      user: "{{ env_var('SNOWFLAKE_USER') }}"
      role: analyst_role
      database: freshcart_dev
      schema: "dev_{{ env_var('DBT_DEV_SCHEMA', 'default') }}"

    ci:
      type: snowflake
      role: ci_service_role
      database: freshcart_ci
      schema: "ci_{{ env_var('PR_NUMBER', 'manual') }}"   # ci_142, ci_143, ...

    prod:
      type: snowflake
      role: pipeline_role
      database: freshcart_prod
      schema: silver   # or gold, depending on the model group

def create_ci_environment(pr_number: int) -> str:
    ci_db = f'freshcart_ci_pr_{pr_number}'
    snowflake_conn.execute(f"""
        CREATE OR REPLACE DATABASE {ci_db}
        CLONE freshcart_prod
        DATA_RETENTION_TIME_IN_DAYS = 1
    """)
    return ci_db

def teardown_ci_environment(pr_number: int) -> None:
    snowflake_conn.execute(f'DROP DATABASE IF EXISTS freshcart_ci_pr_{pr_number}')

# In the CI pipeline: create on PR open → dbt test --target ci → teardown on PR close

$ python -c "from scripts.ci import create_ci_environment; create_ci_environment(142)"
Creating database freshcart_ci_pr_142 as a clone of freshcart_prod...
Done in 2.1s.  Storage cost: $0.00 (shared pages with freshcart_prod)

name: dbt CI
on:
  pull_request:
    branches: [main]
    paths: ['dbt/**']

jobs:
  dbt-ci:
    runs-on: ubuntu-latest
    timeout-minutes: 30
    env:
      CI_SNOWFLAKE_USER: ${{ secrets.CI_SNOWFLAKE_USER }}
      CI_SNOWFLAKE_PASSWORD: ${{ secrets.CI_SNOWFLAKE_PASSWORD }}
      PR_NUMBER: ${{ github.event.pull_request.number }}
    steps:
      - uses: actions/checkout@v4
        with: { fetch-depth: 0 }   # needed for dbt --select state:modified
      - uses: actions/setup-python@v5
        with: { python-version: '3.11', cache: pip }
      - run: pip install dbt-snowflake==1.8.0 dbt-utils
      - name: Create CI database (Zero-Copy Clone)
        run: python scripts/ci/create_ci_db.py --pr ${{ github.event.pull_request.number }}
      - run: dbt deps
        working-directory: dbt

      - name: dbt compile (catch SQL syntax errors)
        working-directory: dbt
        run: dbt compile --target ci

      - name: dbt run — SLIM CI (only changed models + downstream)
        working-directory: dbt
        run: dbt run --target ci --select state:modified+ --defer --state ./prod_artifacts

      - name: dbt test — tests for changed models + downstream
        working-directory: dbt
        run: dbt test --target ci --select state:modified+ --defer --state ./prod_artifacts --store-failures

      - name: Check for breaking schema changes
        run: python scripts/ci/check_schema_changes.py --pr ${{ github.event.pull_request.number }}

      - name: Teardown CI database
        if: always()
        run: python scripts/ci/teardown_ci_db.py --pr ${{ github.event.pull_request.number }}

✓ dbt compile        12s
✓ dbt run (4 models) 48s   ← slim CI: 4 of 150 models
✓ dbt test (4 models) 22s
✓ schema change check 3s   — no breaking changes detected
✓ teardown             4s
Total: 1m 29s

PR changes: silver.orders

state:modified+  selects:
  silver.orders          ← changed directly
  gold.daily_revenue     ← downstream of silver.orders
  gold.customer_ltv      ← downstream of silver.orders
  gold.fct_orders_wide   ← downstream of silver.orders
Skips: silver.customers, silver.payments, and all unrelated gold models.
Runs 4 models instead of 150. CI time: ~1 min instead of 45 min.

# Without --defer:
#   silver.orders → tries freshcart_ci_pr_142.bronze.orders → NOT FOUND → error

# With --defer --state ./prod_artifacts:
#   silver.orders → reads freshcart_prod.bronze.orders → works

# prod_artifacts/manifest.json is the reference — kept current in S3:
aws s3 cp s3://freshcart-ci-artifacts/dbt/manifest.json ./prod_artifacts/   # at CI start
aws s3 cp ./target/manifest.json s3://freshcart-ci-artifacts/dbt/          # after every prod deploy

def detect_breaking_changes(current_manifest: dict, prod_manifest: dict) -> list[str]:
    breaking = []
    for node_id, node in prod_manifest['nodes'].items():
        if node_id not in current_manifest['nodes']:
            breaking.append(f"Model removed: {node['name']}")
            continue
        prod_cols = {c: v['data_type'] for c, v in node.get('columns', {}).items()}
        current_cols = {c: v['data_type'] for c, v in current_manifest['nodes'][node_id].get('columns', {}).items()}
        for col, dtype in prod_cols.items():
            if col not in current_cols:
                breaking.append(f"{node['name']}.{col} removed")
            elif current_cols[col] != dtype:
                breaking.append(f"{node['name']}.{col}: {dtype} → {current_cols[col]}")
    return breaking

BREAKING SCHEMA CHANGES DETECTED:
  - gold.daily_revenue.net_revenue removed
If this is intentional, update all downstream consumers first.

on:
  push:
    branches: [main]

jobs:
  deploy:
    steps:
      - uses: actions/checkout@v4
      - run: |
          dbt deps
          dbt run --target prod
          dbt test --target prod
      - run: aws s3 cp ./target/manifest.json s3://freshcart-ci-artifacts/dbt/

def blue_green_deploy_gold_model(model_name: str, run_date: str):
    # Step 1: build in a shadow schema — not live to analysts yet
    subprocess.run(['dbt', 'run', '--target', 'prod', '--select', model_name,
        '--vars', json.dumps({'run_date': run_date, 'target_schema': 'gold_shadow'})], check=True)

    # Step 2: test the shadow schema before anyone sees it
    subprocess.run(['dbt', 'test', '--target', 'prod', '--select', model_name,
        '--vars', json.dumps({'target_schema': 'gold_shadow'})], check=True)

    # Step 3: atomic swap — analysts see the new version immediately
    conn.execute("BEGIN;")
    conn.execute("ALTER SCHEMA freshcart_prod.gold RENAME TO freshcart_prod.gold_old_20260317;")
    conn.execute("ALTER SCHEMA freshcart_prod.gold_shadow RENAME TO freshcart_prod.gold;")
    conn.execute("COMMIT;")   # both renames atomic — never a window with no 'gold' schema

    # Step 4: keep the old schema for 24h, then drop it
    schedule_schema_drop('gold_old_20260317', delay_hours=24)

-- Step 1: add the column as nullable (this dbt run) — analysts see NULL, no breakage
-- Step 2: backfill as a SEPARATE job, so it doesn't lock the table for 3 hours
UPDATE silver.orders SET tip_amount = 0.0
WHERE tip_amount IS NULL AND created_at < '2026-03-17';   -- rows before the feature launch
-- Step 3: only once backfill is complete, add not_null to schema.yml

# Used by Cloud Composer, MWAA, Astronomer:
# push to main → CI passes → Git Sync detects the change → scheduler re-parses → live

# RISK: no staging step for Airflow DAGs.
#   A syntax error makes the DAG disappear from the UI entirely.
#   A schedule change takes effect immediately — possibly mid-run.

# MITIGATION:
#   python -m py_compile dags/*.py    — catch syntax errors before merge
#   airflow dags list-import-errors   — catch import errors before merge
#   pause the DAG for genuinely risky changes: pause → deploy → verify → unpause

# RISKY: modifying the existing DAG's schedule mid-stream
# DAG('freshcart_morning_pipeline', schedule='0 2 * * *', ...) → schedule='0 6 * * *'
# a run already in progress sees the new schedule on its next evaluation

# SAFER: version the DAG ID
DAG('freshcart_morning_pipeline_v2', schedule='0 6 * * *', ...)
# v1 finishes its current cycle undisturbed; v2 starts fresh on the new schedule
# once v1 has no more in-progress runs, delete it

- run: flake8 dags/ --max-line-length=120
- run: |
    for f in dags/*.py; do python -m py_compile "$f" && echo "OK: $f"; done
- run: airflow db init && airflow dags list-import-errors
- run: python scripts/ci/validate_dag_structure.py   # unique task IDs, no cycles, start/end present
- run: pytest tests/dags/ -v

from airflow.models import DagBag

def test_freshcart_pipeline_dag_structure():
    dagbag = DagBag(dag_folder='dags/', include_examples=False)
    dag = dagbag.get_dag('freshcart_morning_pipeline')
    assert dag is not None, "DAG not found"
    assert len(dagbag.import_errors) == 0, f"Import errors: {dagbag.import_errors}"
    task_ids = [t.task_id for t in dag.tasks]
    assert 'dbt_silver' in task_ids and 'dbt_gold' in task_ids

def test_freshcart_pipeline_task_order():
    dag = DagBag(dag_folder='dags/').get_dag('freshcart_morning_pipeline')
    silver, gold = dag.get_task('dbt_silver'), dag.get_task('dbt_gold')
    assert gold.task_id in [t.task_id for t in silver.downstream_list]

def test_schedule_is_set():
    dag = DagBag(dag_folder='dags/').get_dag('freshcart_morning_pipeline')
    assert dag.schedule_interval is not None
    assert dag.catchup is False, "catchup must be False in production DAGs"

$ pytest tests/dags/ -v
test_freshcart_pipeline_dag_structure PASSED
test_freshcart_pipeline_task_order PASSED
test_schedule_is_set PASSED
========================== 3 passed in 0.41s ===========================

End-to-end   (full pipeline, prod-like data, validated outputs)   ← most valuable, slowest
Integration  (dbt tests against real data volumes)                 ← good coverage, medium speed
Unit         (pure Python — validators, hash key generators)       ← least valuable alone, fastest

Do NOT try to unit-test SQL by mocking the database — that doesn't work.

from pipeline.validate import validate_order_row

def test_valid_order_passes():
    row = {'order_id': 9284751, 'customer_id': 4201938, 'order_amount': 380.00, 'status': 'delivered'}
    result = validate_order_row(row)
    assert result.is_valid, f"Expected valid, got: {result.error}"

def test_negative_amount_rejected():
    row = {'order_id': 1, 'customer_id': 1, 'order_amount': -10, 'status': 'placed'}
    result = validate_order_row(row)
    assert not result.is_valid and result.error_type == 'negative_amount'

def test_hash_key_is_deterministic():
    from pipeline.vault import compute_hub_hk
    assert compute_hub_hk('4201938') == compute_hub_hk('4201938')
    assert compute_hub_hk('ST001') == compute_hub_hk(' st001 ')   # normalised before hashing

$ dbt test --target ci --select state:modified+
FAIL not_null_silver_orders_customer_id (12 rows)
# passed locally on a 1,000-row dev sample — these 12 nulls only exist in production

def test_morning_pipeline_e2e(snowflake_conn, dbt_runner):
    test_date = date.today() - timedelta(days=1)
    result = dbt_runner.run(select='staging.* silver.* gold.*', vars={'run_date': str(test_date)}, target='ci')
    assert result.success, f"Pipeline failed: {result.errors}"

    rows = snowflake_conn.execute(
        f"SELECT COUNT(*) FROM ci_pr_142.gold.daily_revenue WHERE order_date = '{test_date}'").scalar()
    assert 40_000 < rows < 100_000, f"Unexpected row count: {rows}"

    negative_revenue = snowflake_conn.execute(
        f"SELECT COUNT(*) FROM ci_pr_142.gold.daily_revenue WHERE net_revenue < 0").scalar()
    assert negative_revenue == 0

    # Bronze = Silver + DLQ — every extracted row is accounted for somewhere
    bronze = snowflake_conn.execute(f"SELECT COUNT(*) FROM ci_pr_142.bronze.orders WHERE _bronze_date = '{test_date}'").scalar()
    silver = snowflake_conn.execute(f"SELECT COUNT(*) FROM ci_pr_142.silver.orders WHERE DATE(created_at) = '{test_date}'").scalar()
    dlq = snowflake_conn.execute(f"SELECT COUNT(*) FROM ci_pr_142.pipeline.dead_letter_queue WHERE run_date = '{test_date}'").scalar()
    assert bronze == silver + dlq, f"Row count mismatch: {bronze} bronze != {silver} silver + {dlq} dlq"

$ pytest tests/e2e/test_morning_pipeline.py -v
test_morning_pipeline_e2e PASSED
# bronze=48234, silver=48222, dlq=12 → 48234 == 48222 + 12 ✓

A data engineer renames net_revenue to revenue_after_discount in gold.daily_revenue for clarity. No dbt tests fail. The PR merges. Three Metabase dashboards querying net_revenue directly break immediately. Finance notices at 09:00.

✓ SQL compiled successfully
✓ dbt tests passed (not_null, unique on order_date, store_id)
✗ No check that net_revenue was removed
✗ No check that Metabase uses net_revenue
✗ No breakage visible until the prod deploy already happened

Detection: 45 min (analyst reports broken dashboard). Fix: 20 min (alias added, redeployed).
Total impact: 1h 5min of broken Finance dashboards in the morning.

Fix 1 — schema change detection on every PR

def check_for_breaking_column_changes():
    prod = load_manifest('./prod_artifacts/manifest.json')
    current = load_manifest('./target/manifest.json')
    changes = detect_breaking_changes(prod['nodes'], current['nodes'])
    if changes:
        print("BREAKING SCHEMA CHANGES DETECTED:")
        for c in changes:
            print(f"  - {c}")
        print("If this is intentional, update all downstream consumers first.")
        sys.exit(1)

Fix 2 — the backward-compatible migration pattern, going forward

-- This PR: both columns exist, nothing breaks
order_amount - discount_amount AS revenue_after_discount,
order_amount - discount_amount AS net_revenue,  -- backward-compat alias

-- Next PR, after all dashboards have migrated: remove the net_revenue alias

Next time a Gold column is renamed:
CI fails with: "BREAKING SCHEMA CHANGES: net_revenue removed"
Developer sees: ["Metabase: Daily Revenue dashboard", "CFO Report export"]
Cannot merge until consumers are updated or the PR adds a backward-compat alias.
Zero production breakages from schema changes since.