Test data pipelines systematically. Covers Great Expectations, dbt tests, data profiling, anomaly detection, schema validation, and building a data quality SLA.
This page describes techniques, tools, and patterns for verifying data correctness and consistency within data pipelines. It applies when you need to ensure that data transformations, ingestion, or movement preserve integrity, detect anomalies, and maintain expectations across systems.
Use SQL queries with explicit column definitions and type checking to validate schema integrity during ingestion or transformation steps.
-- Check for missing columns in a table
SELECT COUNT(*) FROM (
SELECT 'id' AS col_name UNION ALL
SELECT 'name' UNION ALL
SELECT 'email'
) expected_cols
LEFT JOIN (
SELECT column_name FROM information_schema.columns
WHERE table_name = 'user_data'
) actual_cols ON expected_cols.col_name = actual_cols.column_name
WHERE actual_cols.column_name IS NULL;-- Validate data types for specific columns
SELECT
CASE WHEN COUNT(*) = 0 THEN 'OK' ELSE 'FAIL' END as schema_check
FROM information_schema.columns
WHERE table_name = 'user_events'
AND column_name IN ('event_id', 'timestamp')
AND data_type IN ('uuid', 'timestamp with time zone');If a schema validation step silently passes when columns are missing, downstream transformations may fail without clear error messages. A common failure mode is assuming a column exists when it's actually NULL or dropped.
Run checks for null values, duplicates, and range violations using SQL or custom scripts.
-- Find rows with null values in critical fields
SELECT COUNT(*) FROM user_table
WHERE user_id IS NULL OR email IS NULL;
-- Identify duplicate records by primary key
SELECT user_id, COUNT(*)
FROM user_table
GROUP BY user_id
HAVING COUNT(*) > 1;# Example check script using Python and pandas
import pandas as pd
df = pd.read_csv('data.csv')
null_count = df.isnull().sum()
if null_count.sum() > 0:
print(f"Found {null_count.sum()} null values")
print(null_count[null_count > 0])The most common silent failure is when a check only validates the current batch and ignores historical data. A job might pass a validation for today's data but leave behind corrupted records from earlier runs.
Use timestamp columns to verify that data streams are updating regularly.
-- Check last update time in a table
SELECT MAX(timestamp_column) as latest_update
FROM table_name;
-- Compare with expected freshness window (e.g., 1 hour)
SELECT
CASE
WHEN MAX(timestamp_column) < NOW() - INTERVAL '1 hour'
THEN 'STALE'
ELSE 'FRESH'
END as freshness_status
FROM table_name;If a freshness check doesn't account for expected delays in data ingestion, it may report false positives. For example, a batch job that runs daily at midnight might be considered stale if it hasn't updated by 11:30 PM.
Define and enforce business rules as part of your data pipeline using SQL constraints or custom validation logic.
-- Example constraint to validate email format
ALTER TABLE user_table
ADD CONSTRAINT valid_email_format
CHECK (email ~ '^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$');
-- Check for valid age range
SELECT COUNT(*) FROM user_table
WHERE age < 0 OR age > 150;# Validate constraints via command line
psql -d mydb -c "
SELECT COUNT(*) FROM user_table
WHERE email NOT LIKE '%@%.%';
"A constraint may appear to pass in development but fail in production due to different data volumes or character encoding. The same validation that succeeds with 100 rows might silently drop records when run against a table with millions of entries.
Use custom logic to validate individual data points against known good values or calculated expectations.
-- Example: Check if computed fields match expected values
SELECT
CASE WHEN SUM(calculated_field) = SUM(expected_field) THEN 'PASS' ELSE 'FAIL' END as validation_result
FROM (
SELECT
id,
revenue * tax_rate AS calculated_tax,
reported_tax
FROM sales_data
) validation_table;-- Verify data consistency across related tables
SELECT COUNT(*)
FROM orders o
JOIN customers c ON o.customer_id = c.id
WHERE o.total_amount != c.total_spent;The most common mistake is validating only the most recent data, ignoring older rows that may have been modified or corrupted. A validation that checks only the last 100 records might miss issues in a table with 10 million rows.
Run statistical analysis to detect unexpected patterns or distributions in your data.
-- Calculate basic statistics for numeric columns
SELECT
column_name,
AVG(value) as mean,
STDDEV(value) as stddev,
MIN(value) as min,
MAX(value) as max
FROM (
SELECT revenue as value FROM sales_table
) stats_query;-- Detect anomalies in data distribution
SELECT
COUNT(*) as row_count,
COUNT(DISTINCT category_id) as unique_categories
FROM product_catalog;A profiling check that doesn't account for expected data changes may flag legitimate variations as anomalies. For example, a sudden increase in the number of categories might be normal during a seasonal update rather than an error.
Verify that transformations produce expected outputs using controlled inputs.
-- Test transformation with known input/output pairs
WITH test_input AS (
SELECT '2023-01-01'::date as input_date,
100 as base_value
),
transformed_data AS (
SELECT
input_date,
base_value,
DATE_TRUNC('month', input_date) as month_start,
base_value * 1.1 as increased_value
FROM test_input
)
SELECT
CASE WHEN month_start = '2023-01-01' AND increased_value = 110 THEN 'PASS' ELSE 'FAIL' END as test_result
FROM transformed_data;If transformations are tested only with a small sample, they may miss edge cases. A function that works correctly for most inputs might silently fail when given NULL values or values at boundaries.
Ensure that data moving between systems maintains integrity and completeness.
-- Compare row counts between source and destination
SELECT
(SELECT COUNT(*) FROM source_table) as source_count,
(SELECT COUNT(*) FROM destination_table) as dest_count,
CASE WHEN
(SELECT COUNT(*) FROM source_table) = (SELECT COUNT(*) FROM destination_table)
THEN 'PASS' ELSE 'FAIL' END as migration_status;-- Validate key fields match between systems
SELECT COUNT(*) FROM (
SELECT id FROM source_table
EXCEPT
SELECT id FROM destination_table
) mismatched_ids;A migration test that only compares counts may miss data corruption. A job that moves 100,000 rows but accidentally duplicates some and omits others will still pass a simple row count check.
Set up scheduled checks to run regularly and alert on failures.
#!/bin/bash
# Example quality check script
set -e
# Check for nulls in critical columns
psql -d mydb -c "
SELECT COUNT(*) FROM user_table
WHERE email IS NULL OR user_id IS NULL;
" | grep -q "0" || {
echo "Critical data validation failed"
exit 1
}
# Check freshness
psql -d mydb -c "
SELECT
CASE WHEN MAX(created_at) < NOW() - INTERVAL '1 hour' THEN 'STALE' ELSE 'FRESH' END as status
FROM user_table;
" | grep -q "FRESH" || {
echo "Data freshness check failed"
exit 1
}A scheduled check may fail to detect issues that occur during the window between checks. A job that runs every 15 minutes might miss failures that happen in the gap between executions.
Implement clear failure handling that distinguishes between recoverable and fatal errors.
-- Example: Insert validation result into a monitoring table
INSERT INTO data_quality_log (check_name, status, timestamp, details)
VALUES (
'email_format_check',
CASE WHEN
(SELECT COUNT(*) FROM user_table WHERE email NOT LIKE '%@%.%') = 0
THEN 'PASS' ELSE 'FAIL' END,
NOW(),
'Found ' || (SELECT COUNT(*) FROM user_table WHERE email NOT LIKE '%@%.%') || ' invalid emails'
);Validation failures that don't write to a log or alert system may be silently ignored. A check that produces output but doesn't signal failure to the orchestrator can make an entire pipeline appear healthy while data is actually corrupted.
The most important consideration is ensuring that every validation step explicitly signals success or failure and logs its results in a way that can be monitored and audited.
This page was rewritten on 15 October 2026. It replaced a templated version that was not specific to its own title. How these pages are written and checked. If anything here is wrong, tell us at [email protected] and we will correct it.
We use cookies for analytics (Google Analytics) and advertising (Google AdSense) to improve your experience and support free content. Privacy Policy