Data Quality — Dimensions, Testing, Monitoring, and Contracts

SOURCE SYSTEM (before ingestion):
  Cost: reject the record, log to DLQ, notify source team
  Recovery time: minutes

BRONZE LAYER (after landing):
  Cost: record in DLQ, Bronze intact, Silver/Gold unaffected
  Recovery time: hours (after source team fixes and resends)

SILVER LAYER (after transformation):
  Cost: dbt run fails, Silver not updated, Gold build blocked
  Recovery time: hours to a day

GOLD LAYER (after aggregation):
  Cost: Gold table has wrong data, dashboards show wrong metrics
  Recovery time: 1-3 days (investigation + fix + rebuild)

ANALYST DASHBOARD (after analyst queries):
  Cost: analyst escalates, business decisions already made on wrong data
  Recovery time: unknown, trust damage lasting weeks

THE RULE: every layer a quality issue traverses multiplies its cost by 10×.

models:
  - name: silver_orders
    columns:
      - name: order_id
        tests:
          - not_null           # catches missing PKs
          - unique              # catches duplicates at the grain

      - name: customer_id
        tests:
          - not_null
          - relationships:      # referential integrity to parent table
              to: ref('silver_customers')
              field: customer_id
              severity: warn    # warn not error: some orders arrive before customers

      - name: status
        tests:
          - not_null
          - accepted_values:    # domain validation
              values: ['placed', 'confirmed', 'preparing', 'ready',
                       'picked_up', 'delivering', 'delivered', 'cancelled']

      - name: order_amount
        tests:
          - not_null
          - dbt_utils.accepted_range: {min_value: 0, max_value: 500000}

models:
  - name: silver_customers
    tests:
      - dbt_utils.equal_rowcount: {compare_model: ref('stg_customers')}
      - dbt_utils.recency: {datepart: hour, field: updated_at, interval: 25}

sources:
  - name: bronze
    database: freshcart_prod
    tables:
      - name: orders
        freshness:
          warn_after:  {count: 25, period: hour}
          error_after: {count: 49, period: hour}
        loaded_at_field: _bronze_date

$ dbt test --select silver_orders
1 of 4 PASS not_null_silver_orders_order_id
2 of 4 PASS unique_silver_orders_order_id
3 of 4 WARN relationships_silver_orders_customer_id__customer_id__ref_silver_customers_
4 of 4 PASS accepted_values_silver_orders_status
Done. PASS=3 WARN=1 ERROR=0

-- tests/assert_no_negative_amounts.sql
-- Passes when this query returns ZERO rows.
SELECT order_id, order_amount
FROM {{ ref('silver_orders') }}
WHERE order_amount < 0;

columns:
  - name: delivered_at
    tests:
      - dbt_utils.expression_is_true:
          expression: "delivered_at >= created_at OR delivered_at IS NULL"

  - name: order_amount
    tests:
      - dbt_utils.expression_is_true:
          expression: "order_amount >= discount_amount"

{% test assert_column_sum_equals(model, column_name, compare_model, compare_column) %}
WITH model_sum   AS (SELECT SUM({{ column_name }}) AS total FROM {{ model }}),
     compare_sum AS (SELECT SUM({{ compare_column }}) AS total FROM {{ compare_model }})
SELECT m.total AS model_total, c.total AS compare_total, ABS(m.total - c.total) AS difference
FROM model_sum m, compare_sum c
WHERE ABS(m.total - c.total) > 0.01
{% endtest %}

# Usage:
# - name: order_amount
#   tests:
#     - assert_column_sum_equals: {compare_model: ref('silver_payments'), compare_column: payment_amount}

$ dbt test -s silver_orders --store-failures
FAIL assert_no_negative_amounts (3 rows)
# creates dbt_test__audit.assert_no_negative_amounts — query it directly:
SELECT * FROM dbt_test__audit.assert_no_negative_amounts;

WITH daily_counts AS (
    SELECT DATE(ingested_at) load_date, COUNT(*) row_count
    FROM silver.orders WHERE ingested_at >= CURRENT_DATE - 30
    GROUP BY 1
),
stats AS (
    SELECT load_date, row_count,
        AVG(row_count) OVER (ORDER BY load_date ROWS BETWEEN 7 PRECEDING AND 1 PRECEDING) rolling_7d_avg
    FROM daily_counts
)
SELECT load_date, row_count, ROUND(rolling_7d_avg, 0) expected_avg,
    CASE WHEN ABS(row_count - rolling_7d_avg) / NULLIF(rolling_7d_avg, 0) > 0.5 THEN 'CRITICAL'
         WHEN ABS(row_count - rolling_7d_avg) / NULLIF(rolling_7d_avg, 0) > 0.3 THEN 'WARNING'
         ELSE 'OK' END AS status
FROM stats WHERE load_date = CURRENT_DATE;

load_date    row_count  expected_avg  status
2026-03-17   4,820      48,200        CRITICAL
# Monday's orders table has 10% of its expected volume — alert fires
# before any analyst opens a dashboard

import statistics

def detect_metric_anomaly(metric_name: str, today_value: float,
                           historical_values: list[float], z_threshold: float = 3.0) -> dict:
    if len(historical_values) < 7:
        return {'status': 'insufficient_history', 'z_score': None}
    mean, stdev = statistics.mean(historical_values), statistics.stdev(historical_values)
    if stdev == 0:
        return {'status': 'no_variance', 'z_score': 0}
    z_score = abs(today_value - mean) / stdev
    return {'metric': metric_name, 'today_value': today_value, 'mean': round(mean, 2),
            'z_score': round(z_score, 2), 'status': 'ANOMALY' if z_score > z_threshold else 'OK'}

result = detect_metric_anomaly('daily_revenue', query_gold_revenue(date='2026-03-17'),
                                query_gold_revenue(last_n_days=30))
if result['status'] == 'ANOMALY':
    send_alert(f"Revenue anomaly: z_score={result['z_score']}, today={result['today_value']}")

>>> detect_metric_anomaly('daily_revenue', 812000.0, [420000]*30)
{'metric': 'daily_revenue', 'today_value': 812000.0, 'mean': 420000.0, 'z_score': 4.1, 'status': 'ANOMALY'}

sources:
  - name: bronze
    tables:
      - name: orders
        loaded_at_field: ingested_at
        freshness:
          warn_after:  {count: 2, period: hour}
          error_after: {count: 6, period: hour}

# dbt source freshness — run as an Airflow task, fail the DAG if stale

# Elementary (pip install elementary-data) auto-tracks, per model:
# row count, null % per column, distinct value count — all per time period,
# and alerts on deviation with no rules written by hand.

import great_expectations as gx
from great_expectations.core.batch import RuntimeBatchRequest
from pathlib import Path
import pandas as pd

context = gx.get_context()

def validate_vendor_file(file_path: str, pipeline_run_id: str) -> bool:
    """Validate a vendor CSV against an expectation suite. Quarantines on failure."""
    df = pd.read_csv(file_path)
    batch_request = RuntimeBatchRequest(
        datasource_name='pandas_datasource', data_connector_name='runtime_data_connector',
        data_asset_name='vendor_deliveries', runtime_parameters={'batch_data': df},
        batch_identifiers={'run_id': pipeline_run_id},
    )
    result = context.run_checkpoint(
        checkpoint_name='vendor_deliveries_checkpoint',
        validations=[{'batch_request': batch_request, 'expectation_suite_name': 'vendor_deliveries.critical'}],
    )
    if not result.success:
        quarantine_path = Path('/data/quarantine') / Path(file_path).name
        Path(file_path).rename(quarantine_path)
        send_alert(f'Vendor file failed validation: {file_path}. Quarantined at: {quarantine_path}.')
        return False
    return True

# suite.add_expectation(gx.expectations.ExpectColumnValuesToNotBeNull(column="delivery_id"))
# suite.add_expectation(gx.expectations.ExpectColumnValuesToBeUnique(column="delivery_id"))
# suite.add_expectation(gx.expectations.ExpectColumnValuesToBeBetween(
#     column="delivery_fee", min_value=0, max_value=5000, mostly=0.99))
# suite.add_expectation(gx.expectations.ExpectTableRowCountToBeBetween(min_value=1000, max_value=500000))

INFO Validating vendor file: shipfast_weekly_2026-03-17.csv
FAIL ExpectColumnValuesToBeBetween(delivery_fee): 3.2% of values exceed max_value=5000
WARNING Vendor file failed validation — quarantined at /data/quarantine/shipfast_weekly_2026-03-17.csv

checks for silver_orders:
  - row_count > 10000:
      name: Minimum row count — pipeline produced data
  - missing_count(order_id) = 0:
      name: No missing order IDs
  - duplicate_count(order_id) = 0:
      name: No duplicate order IDs
  - invalid_count(status) = 0:
      name: All statuses are valid
      valid values: [placed, confirmed, preparing, ready, picked_up, delivering, delivered, cancelled]
  - min(order_amount) >= 0:
      name: No negative order amounts
  - freshness(updated_at) < 2h:
      name: Data is less than 2 hours old

def run_soda_checks(**context):
    from soda.scan import Scan
    scan = Scan()
    scan.set_data_source_name('freshcart_snowflake')
    scan.add_sodacl_yaml_files(path='checks/silver_orders.yml')
    scan.execute()
    if scan.has_error_logs():
        raise ValueError(f'Soda checks failed: {scan.get_error_count()} errors.')

quality_check_task = PythonOperator(task_id='soda_silver_orders', python_callable=run_soda_checks)
dbt_silver_task >> quality_check_task >> dbt_gold_task   # Gold only runs if checks pass

id: orders_api_v2
version: 2.3.1
owner: orders-team@freshcart.com
consumer: data-engineering@freshcart.com

sla:
  schedule: "every 15 minutes"
  latency_sla: "data available within 5 minutes of order event"

schema:
  fields:
    - name: order_id
      type: integer
      required: true
      unique: true
    - name: order_amount
      type: decimal(10, 2)
      required: true
      constraints: {min: 0, max: 500000}
    - name: status
      type: string
      required: true
      allowed_values: [placed, confirmed, preparing, ready, picked_up, delivering, delivered, cancelled]

quality:
  completeness: ["order_id is never null", "row_count is within ±20% of 7-day rolling average"]
  timeliness: ["data delivered within 5 minutes of event"]
  schema_changes:
    breaking_change_notice: "30 days minimum before any breaking change"
    additive_change_notice: "7 days minimum before adding new fields"

from dataclasses import dataclass
from typing import Any
import yaml

@dataclass
class ContractViolation:
    field: str
    constraint: str
    actual_value: Any
    severity: str

def validate_against_contract(df, contract_path: str) -> list[ContractViolation]:
    """Returns list of violations. Empty list = contract satisfied."""
    with open(contract_path) as f:
        contract = yaml.safe_load(f)
    violations = []

    for field_spec in contract['schema']['fields']:
        name = field_spec['name']
        if field_spec.get('required') and name not in df.columns:
            violations.append(ContractViolation(name, 'required_field_missing', None, 'error'))
            continue
        if field_spec.get('required'):
            nulls = df[name].isna().sum()
            if nulls > 0:
                violations.append(ContractViolation(name, 'not_null', nulls, 'error'))
        if 'allowed_values' in field_spec:
            invalid = df[name].dropna()[~df[name].dropna().isin(field_spec['allowed_values'])]
            if len(invalid) > 0:
                violations.append(ContractViolation(name, 'allowed_values', invalid.unique().tolist()[:5], 'error'))

    return violations

>>> validate_against_contract(bronze_orders_df, 'contracts/orders_api_v2.yml')
[ContractViolation(field='status', constraint='allowed_values',
                    actual_value=['scheduled'], severity='error')]
# exactly the violation from this module's Real World section below

def is_breaking_change(old_schema: dict, new_schema: dict) -> list[str]:
    breaking = []
    old_fields = {f['name']: f for f in old_schema['schema']['fields']}
    new_fields = {f['name']: f for f in new_schema['schema']['fields']}

    for name in old_fields:
        if name not in new_fields:
            breaking.append(f"Field '{name}' removed — consumers may break")

    for name, spec in new_fields.items():
        if name not in old_fields and spec.get('required'):
            breaking.append(f"New required field '{name}' added — existing data invalid")

    for name in old_fields:
        if name in new_fields and old_fields[name]['type'] != new_fields[name]['type']:
            breaking.append(f"Field '{name}' type changed: {old_fields[name]['type']} → {new_fields[name]['type']}")

    for name in old_fields:
        old_allowed = set(old_fields.get(name, {}).get('allowed_values', []))
        new_allowed = set(new_fields.get(name, {}).get('allowed_values', []))
        if old_allowed and new_allowed and not new_allowed.issuperset(old_allowed):
            breaking.append(f"Field '{name}': allowed values {old_allowed - new_allowed} removed")

    return breaking

$ python validate_contract_backwards_compatible.py --old v2.2.0.yml --new v2.3.0.yml
Field 'status': breaking change NOT detected (new value 'scheduled' only ADDS an option)
✓ Additive change — requires 7-day notice, not 30. PR may proceed.
# had this check existed, it's exactly what should have caught the enum
# addition described in this module's Real World section

CREATE TABLE monitoring.data_quality_results (
    check_id      UUID DEFAULT gen_random_uuid() PRIMARY KEY,
    run_id        UUID NOT NULL,
    table_name    VARCHAR(200) NOT NULL,
    check_name    VARCHAR(200) NOT NULL,
    check_type    VARCHAR(50)  NOT NULL,   -- 'dbt_test', 'soda', 'custom', 'anomaly'
    status        VARCHAR(10)  NOT NULL,   -- 'pass', 'fail', 'warn'
    failure_count BIGINT,
    failure_rate  DECIMAL(6,4),
    message       TEXT,
    checked_at    TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_dq_status_date ON monitoring.data_quality_results (status, checked_at)
    WHERE status IN ('fail', 'warn');

-- Daily pass rate per table
SELECT table_name, DATE(checked_at) check_date,
    ROUND(SUM(CASE WHEN status='pass' THEN 1 ELSE 0 END)::NUMERIC / COUNT(*) * 100, 1) pass_rate_pct
FROM monitoring.data_quality_results
WHERE checked_at >= CURRENT_DATE - 30 GROUP BY 1, 2 ORDER BY 2 DESC;

-- Is quality improving or degrading week over week?
WITH weekly AS (
    SELECT DATE_TRUNC('week', checked_at) week_start, table_name,
           SUM(CASE WHEN status='fail' THEN 1 ELSE 0 END) failures
    FROM monitoring.data_quality_results WHERE checked_at >= CURRENT_DATE - 90 GROUP BY 1, 2
)
SELECT week_start, table_name, failures,
    failures - LAG(failures) OVER (PARTITION BY table_name ORDER BY week_start) week_over_week_change
FROM weekly ORDER BY week_start DESC;

-- Tables failing right now, worst first
SELECT table_name, check_name, failure_rate, message
FROM monitoring.data_quality_results
WHERE DATE(checked_at) = CURRENT_DATE AND status = 'fail' AND severity = 'error'
ORDER BY failure_rate DESC;

table_name       check_date   pass_rate_pct
silver_orders    2026-03-17   97.1
silver_customers 2026-03-17   100.0
silver_payments  2026-03-17   84.3   ← worth a look

with DAG('freshcart_morning_pipeline', ...) as dag:

    extract_orders = PythonOperator(task_id='extract_orders', python_callable=run_extraction)

    bronze_quality = BashOperator(
        task_id='bronze_quality_check',
        bash_command='dbt source freshness --select source:bronze.orders',
    )

    dbt_silver = BashOperator(task_id='dbt_silver',
        bash_command='dbt run --select staging.* silver.* --vars \'{"run_date": "{{ ds }}"}\'')

    silver_tests = BashOperator(task_id='silver_quality_tests',
        bash_command='dbt test --select silver.* --store-failures')

def soda_silver_check(**context):
    from soda.scan import Scan
    scan = Scan()
    scan.set_data_source_name('freshcart_snowflake')
    scan.add_sodacl_yaml_files(path='checks/silver_orders.yml')
    scan.execute()
    write_soda_results_to_monitoring(scan, context['run_id'])
    if scan.has_error_logs():
        raise ValueError('Soda anomaly check failed for Silver orders')

silver_anomaly = PythonOperator(task_id='silver_anomaly_check', python_callable=soda_silver_check)
dbt_gold  = BashOperator(task_id='dbt_gold', bash_command='dbt run --select gold.*')
gold_tests = BashOperator(task_id='gold_quality_tests', bash_command='dbt test --select gold.*')

def post_pipeline_quality_report(**context):
    result = query_quality_results(date=context['ds'])
    send_slack_message(channel='#data-quality',
        text=f'Pipeline quality: {result.pass_rate}% checks passed. {result.total_failures} failures.')

quality_report = PythonOperator(task_id='quality_report', python_callable=post_pipeline_quality_report,
                                 trigger_rule='all_done')   # runs whether upstream passed or failed

(extract_orders >> bronze_quality >> dbt_silver >> silver_tests
 >> silver_anomaly >> dbt_gold >> gold_tests >> quality_report)

Graph view — freshcart_morning_pipeline
extract_orders → bronze_quality → dbt_silver → silver_tests → silver_anomaly
              → dbt_gold → gold_tests → quality_report
Slack: "Pipeline quality: 97.1% checks passed. 3 failures."

The orders application team added a new status value — “scheduled” — for a new pre-order feature, deployed Friday evening without notifying data engineering. By Monday, 12,847 orders with status='scheduled' were rejected from Silver by the accepted_values test and sitting in the DLQ. The finance dashboard showed no pre-order revenue. An analyst noticed Tuesday.

-- STEP 1: check Silver dbt test failures since Friday
SELECT run_id, check_name, failure_count, message FROM monitoring.data_quality_results
WHERE table_name = 'silver_orders' AND status = 'fail' AND checked_at >= '2026-03-14';
-- 47 runs, all: accepted_values_silver_orders_status  "Values not in set: ['scheduled']"
-- ~600,000 rows total rejected across the 47 runs

-- STEP 2: confirm the root cause
SELECT DISTINCT status FROM bronze.orders WHERE _bronze_date >= '2026-03-14';
-- placed, confirmed, delivering, delivered, cancelled, scheduled ← new

-- STEP 3: quantify impact
SELECT SUM(order_amount) FROM bronze.orders WHERE status = 'scheduled';
-- $4.82 million unloaded to Silver/Gold

# a) Update VALID_STATUSES in pipeline/validate.py to include 'scheduled'
# b) Update dbt schema.yml accepted_values to include 'scheduled'
# c) Bump the data contract version: contracts/orders_api_v2.yml
$ python dlq_reprocess.py --pipeline orders_incremental --start-date 2026-03-14 --force-reload

DLQ reprocessing complete: attempted=598234 reprocessed=598234 failed=0

SELECT COUNT(*) FROM silver.orders WHERE status = 'scheduled';
-- 598,234 ← all reprocessed correctly

Total impact: data missing from Silver/Gold for 2 days 14 hours, a $4.82 million revenue gap in dashboards for 67 hours. The incident was caught by dbt’s accepted_values test exactly as designed — the failure was in process, not tooling: no data contract enforcement meant the orders team had no way to know their enum change would break the downstream pipeline. Going forward, the contract now requires 30-day notice for enum changes, a CI check blocks unreviewed allowed_values additions, and Elementary was added for automated anomaly detection — the Z-score check would have caught this Friday evening, not Tuesday.