BigQuery Data Pipeline Validation: Testing SQL Transforms, Data Quality, and Pipeline Correctness

HelpMeTest

07 Aug 2026 — 6 min read

BigQuery data pipelines have a testing problem. The transformations run in SQL, the data volumes are large, and a broken pipeline often produces plausible-looking results that fail to reveal themselves as wrong until someone checks a business metric two weeks later.

This guide covers practical validation strategies for BigQuery pipelines: testing SQL transforms, verifying data quality, validating schemas, and running integration tests that catch pipeline regressions before they affect production data.

The Core Validation Problem

A BigQuery pipeline that joins three tables, applies filters, and aggregates results can be wrong in many ways that don't produce errors:

A JOIN condition that's off by one field produces wrong join results
A date filter that uses < instead of <= silently drops a day's data
A GROUP BY that's missing a column produces incorrect aggregation
A COALESCE that handles NULL incorrectly produces unexpected values

None of these produce query errors. They produce incorrect data. The only way to catch them is testing.

Unit Testing SQL Transforms with dbt

If you use dbt (data build tool) for transformations, you get a testing framework built in. For raw BigQuery SQL, you can apply the same principles manually.

dbt Schema Tests

The fastest way to validate BigQuery data:

# models/schema.yml
version: 2

models:
  - name: orders
    description: "Cleaned and enriched orders data"
    columns:
      - name: order_id
        description: "Unique order identifier"
        tests:
          - unique
          - not_null

      - name: customer_id
        tests:
          - not_null
          - relationships:
              to: ref('customers')
              field: customer_id

      - name: status
        tests:
          - accepted_values:
              values: ['pending', 'processing', 'shipped', 'delivered', 'cancelled']

      - name: total_amount
        tests:
          - not_null
          - dbt_expectations.expect_column_values_to_be_between:
              min_value: 0
              max_value: 1000000

      - name: created_at
        tests:
          - not_null
          - dbt_expectations.expect_column_values_to_be_of_type:
              column_type: timestamp

Run tests:

dbt test --models orders
dbt test --models orders --select test_type:singular

Custom dbt Singular Tests

For business logic that generic tests can't cover:

-- tests/assert_total_matches_line_items.sql
-- Fails if order totals don't match the sum of their line items

SELECT
  o.order_id,
  o.total_amount AS header_total,
  SUM(li.quantity * li.unit_price) AS calculated_total,
  ABS(o.total_amount - SUM(li.quantity * li.unit_price)) AS discrepancy
FROM {{ ref('orders') }} o
JOIN {{ ref('order_line_items') }} li USING (order_id)
GROUP BY 1, 2
HAVING ABS(header_total - calculated_total) > 0.01

If this query returns any rows, the test fails. The threshold (0.01) accounts for floating point arithmetic.

Testing SQL Transforms Directly in BigQuery

For pipelines not using dbt, test SQL transforms by:

Creating small reference datasets
Running the transform SQL
Comparing output to expected results

# test_pipeline_transforms.py
from google.cloud import bigquery
import pytest

client = bigquery.Client(project='test-project')
DATASET = 'pipeline_tests'

def run_query(sql: str) -> list[dict]:
    """Execute SQL and return results as list of dicts."""
    return [dict(row) for row in client.query(sql).result()]

@pytest.fixture(scope='module', autouse=True)
def setup_test_tables():
    """Create test tables with known data."""
    # Create orders table
    client.query(f"""
        CREATE OR REPLACE TABLE `{DATASET}.raw_orders` AS
        SELECT * FROM UNNEST([
            STRUCT('ord-001' AS order_id, 'cust-1' AS customer_id, 100.00 AS amount, 'USD' AS currency, TIMESTAMP '2024-01-15' AS created_at),
            STRUCT('ord-002', 'cust-2', 250.00, 'EUR', TIMESTAMP '2024-01-16'),
            STRUCT('ord-003', 'cust-1', 75.00, 'USD', TIMESTAMP '2024-01-17'),
            STRUCT('ord-004', 'cust-3', 0.00, 'USD', TIMESTAMP '2024-01-18'),  -- Zero amount edge case
            STRUCT('ord-005', NULL, 50.00, 'USD', TIMESTAMP '2024-01-19'),     -- Null customer edge case
        ])
    """).result()

    # Create exchange rates table
    client.query(f"""
        CREATE OR REPLACE TABLE `{DATASET}.exchange_rates` AS
        SELECT * FROM UNNEST([
            STRUCT('EUR' AS currency, 1.08 AS rate_to_usd, DATE '2024-01-16' AS rate_date),
            STRUCT('USD', 1.00, DATE '2024-01-16'),
        ])
    """).result()

def test_currency_normalization_converts_eur_to_usd():
    """The normalization transform should convert all amounts to USD."""
    result = run_query(f"""
        SELECT order_id, amount_usd
        FROM `{DATASET}.normalized_orders`
        WHERE order_id = 'ord-002'
    """)

    assert len(result) == 1
    assert abs(result[0]['amount_usd'] - 270.00) < 0.01  # 250 EUR * 1.08

def test_null_customer_orders_are_excluded():
    """Orders without customer IDs should be filtered from the output."""
    result = run_query(f"""
        SELECT order_id FROM `{DATASET}.normalized_orders`
        WHERE order_id = 'ord-005'
    """)

    assert len(result) == 0, "Null customer orders should be excluded"

def test_zero_amount_orders_are_included():
    """Zero-amount orders (refunds, samples) should be preserved."""
    result = run_query(f"""
        SELECT order_id, amount_usd
        FROM `{DATASET}.normalized_orders`
        WHERE order_id = 'ord-004'
    """)

    assert len(result) == 1
    assert result[0]['amount_usd'] == 0.00

def test_no_duplicate_order_ids():
    """Each order should appear exactly once after normalization."""
    result = run_query(f"""
        SELECT order_id, COUNT(*) AS cnt
        FROM `{DATASET}.normalized_orders`
        GROUP BY order_id
        HAVING cnt > 1
    """)

    assert len(result) == 0, f"Found duplicate order IDs: {result}"

def test_aggregation_totals_are_correct():
    """Daily totals should match sum of individual orders."""
    result = run_query(f"""
        WITH individual AS (
            SELECT DATE(created_at) AS order_date, SUM(amount_usd) AS total
            FROM `{DATASET}.normalized_orders`
            GROUP BY 1
        ),
        aggregated AS (
            SELECT order_date, daily_total
            FROM `{DATASET}.daily_order_totals`
        )
        SELECT i.order_date, i.total AS expected, a.daily_total AS actual
        FROM individual i
        JOIN aggregated a USING (order_date)
        WHERE ABS(i.total - a.daily_total) > 0.01
    """)

    assert len(result) == 0, f"Aggregation totals don't match: {result}"

Data Quality Monitoring

Beyond correctness testing, production pipelines need continuous quality monitoring:

-- data_quality_checks.sql
-- Run this query and alert if any check fails (returns rows)

WITH checks AS (
  -- Check 1: Row count didn't drop more than 10% from yesterday
  SELECT
    'row_count_drop' AS check_name,
    CASE
      WHEN today_count < yesterday_count * 0.9
      THEN 'FAIL: Row count dropped by more than 10%'
      ELSE 'PASS'
    END AS status,
    today_count,
    yesterday_count
  FROM (
    SELECT
      COUNTIF(DATE(created_at) = CURRENT_DATE()) AS today_count,
      COUNTIF(DATE(created_at) = DATE_SUB(CURRENT_DATE(), INTERVAL 1 DAY)) AS yesterday_count
    FROM `prod.orders`
  )

  UNION ALL

  -- Check 2: No null values in critical columns
  SELECT
    'null_order_ids' AS check_name,
    CASE
      WHEN null_count > 0 THEN CONCAT('FAIL: ', CAST(null_count AS STRING), ' null order_ids found')
      ELSE 'PASS'
    END AS status,
    null_count,
    0 AS yesterday_count
  FROM (
    SELECT COUNTIF(order_id IS NULL) AS null_count FROM `prod.orders`
    WHERE DATE(created_at) = CURRENT_DATE()
  )

  UNION ALL

  -- Check 3: All amounts are positive
  SELECT
    'negative_amounts' AS check_name,
    CASE
      WHEN negative_count > 0 THEN CONCAT('FAIL: ', CAST(negative_count AS STRING), ' negative amounts')
      ELSE 'PASS'
    END AS status,
    negative_count,
    0 AS yesterday_count
  FROM (
    SELECT COUNTIF(amount < 0) AS negative_count FROM `prod.orders`
    WHERE DATE(created_at) = CURRENT_DATE()
  )
)

SELECT * FROM checks WHERE status != 'PASS'

# quality_monitor.py — Run in Cloud Scheduler or as a Cloud Function
from google.cloud import bigquery
import subprocess

def run_quality_checks():
    client = bigquery.Client()

    with open('data_quality_checks.sql') as f:
        sql = f.read()

    results = list(client.query(sql).result())

    if results:
        failures = [dict(r) for r in results]
        raise RuntimeError(f"Data quality checks failed: {failures}")

    print("All data quality checks passed.")

Schema Validation

BigQuery schema changes (added/removed columns, type changes) can silently break downstream consumers:

# test_schema_stability.py
from google.cloud import bigquery
import pytest

EXPECTED_SCHEMA = {
    'orders': {
        'order_id': 'STRING',
        'customer_id': 'STRING',
        'total_amount': 'FLOAT64',
        'currency': 'STRING',
        'status': 'STRING',
        'created_at': 'TIMESTAMP',
    }
}

def test_orders_table_schema_unchanged():
    client = bigquery.Client()
    table = client.get_table('prod.orders')

    actual = {field.name: field.field_type for field in table.schema}

    for col, expected_type in EXPECTED_SCHEMA['orders'].items():
        assert col in actual, f"Column '{col}' is missing from orders table"
        assert actual[col] == expected_type, (
            f"Column '{col}' type changed: expected {expected_type}, got {actual[col]}"
        )

def test_no_unexpected_columns_removed():
    """Alert if expected columns disappear."""
    client = bigquery.Client()
    table = client.get_table('prod.orders')
    actual_columns = {field.name for field in table.schema}

    expected_columns = set(EXPECTED_SCHEMA['orders'].keys())
    missing = expected_columns - actual_columns

    assert not missing, f"Columns removed from orders table: {missing}"

Pipeline Integration Testing End-to-End

For testing the full pipeline (source → transform → destination):

# test_pipeline_e2e.py
import pytest
import time
from google.cloud import bigquery, storage, pubsub_v1
import json

def test_full_pipeline_processes_event():
    """
    End-to-end: publish an event → verify it appears in BigQuery within 60s.
    Requires a deployed pipeline (Dataflow or Cloud Functions).
    """
    client = bigquery.Client()

    # Publish test event
    publisher = pubsub_v1.PublisherClient()
    topic = publisher.topic_path('my-project', 'orders-events')

    test_order = {
        'order_id': f'e2e-test-{int(time.time())}',
        'customer_id': 'test-customer',
        'total_amount': 123.45,
        'currency': 'USD',
        'created_at': '2024-01-20T10:00:00Z',
    }

    publisher.publish(topic, json.dumps(test_order).encode()).result()

    # Wait for pipeline to process (adjust timeout based on your pipeline latency)
    deadline = time.time() + 60
    while time.time() < deadline:
        result = list(client.query(f"""
            SELECT order_id, total_amount
            FROM `my-project.pipeline_output.orders`
            WHERE order_id = '{test_order["order_id"]}'
        """).result())

        if result:
            assert result[0]['total_amount'] == pytest.approx(123.45, abs=0.01)
            return  # Test passed

        time.sleep(5)

    pytest.fail(f"Order {test_order['order_id']} did not appear in BigQuery within 60 seconds")

CI Configuration

# .github/workflows/bigquery-pipeline.yml
name: BigQuery Pipeline Tests

on:
  push:
    paths:
      - 'pipeline/**'
      - 'models/**'
      - 'tests/**'

jobs:
  unit-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

      - run: pip install dbt-bigquery pytest google-cloud-bigquery

      - name: Run dbt tests (against BigQuery test dataset)
        run: dbt test --target test
        env:
          GOOGLE_APPLICATION_CREDENTIALS: ${{ secrets.GCP_SA_KEY_PATH }}

      - name: Run SQL transform tests
        run: pytest tests/unit/ -v -k "not e2e"
        env:
          GOOGLE_CLOUD_PROJECT: ${{ secrets.GCP_TEST_PROJECT }}

  e2e-tests:
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    needs: unit-tests
    steps:
      - uses: actions/checkout@v4

      - run: pip install pytest google-cloud-bigquery google-cloud-pubsub

      - name: Run end-to-end pipeline tests
        run: pytest tests/ -k "e2e" -v --timeout=120
        env:
          GOOGLE_CLOUD_PROJECT: ${{ secrets.GCP_STAGING_PROJECT }}

The Testing Hierarchy for BigQuery Pipelines

SQL unit tests (fastest): Test individual transforms with small, controlled datasets. Catch logic bugs.
Schema tests: Verify column types and nullability. Catch structure changes.
Data quality checks: Run continuously against production data. Catch data drift.
Integration tests: Test the full pipeline end-to-end. Catch wiring and configuration issues.

Most teams run only the first layer. Adding data quality monitoring catches 80% of production issues before users notice them — and it's usually just a scheduled SQL query.

HelpMeTest can monitor the output of your BigQuery pipelines by running behavioral checks against applications that consume them, alerting you when pipeline changes break downstream functionality. Start free →