---
name: data-validation
description: |
  Use this skill when the user wants to verify data parity between two
  databases during a migration. Triggers on: "validate migration",
  "check data parity", "compare old and new", "are counts matching",
  "verify backfill", "check for drift", "reconcile databases".

  Generates: validation queries, comparison scripts, drift reports.

  Do NOT use for: greenfield data validation (use schema constraints),
  backup verification (separate concern), or testing new code (use
  test-generator skill).
---

# Data Validation

You generate validation queries and comparison logic to verify data parity
between two databases during migration. You catch the bugs that simple
COUNT(*) won't catch.

## Validation hierarchy

Run validations in this order. Earlier checks are cheaper and rule out
broad issues; later checks catch subtle bugs.

### Level 1: Row counts

The dumbest, fastest check. If counts don't match, stop everything else.

```sql
-- Source (SQL Server)
SELECT COUNT(*) FROM customers;

-- Target (Postgres)
SELECT COUNT(*) FROM customers;

-- Source
SELECT
  COUNT(*) AS total_count,
  SUM(amount) AS total_amount,
  AVG(amount) AS avg_amount,
  MIN(created_at) AS earliest,
  MAX(created_at) AS latest,
  COUNT(DISTINCT customer_id) AS unique_customers
FROM orders;

-- Target
SELECT
  COUNT(*) AS total_count,
  SUM(amount) AS total_amount,
  AVG(amount) AS avg_amount,
  MIN(created_at) AS earliest,
  MAX(created_at) AS latest,
  COUNT(DISTINCT customer_id) AS unique_customers
FROM orders;

-- Source
SELECT status, COUNT(*) FROM orders GROUP BY status;

-- Target
SELECT status, COUNT(*) FROM orders GROUP BY status;

-- Orphaned orders (customer_id pointing to non-existent customer)
SELECT COUNT(*)
FROM orders o
LEFT JOIN customers c ON o.customer_id = c.id
WHERE c.id IS NULL;

-- Source: 50 random orders
SELECT TOP 50 * FROM orders ORDER BY NEWID();

-- Source: orders created in last hour
SELECT id, created_at FROM orders
WHERE created_at >= DATEADD(hour, -1, GETUTCDATE())
ORDER BY created_at;

-- Target: same query
SELECT id, created_at FROM orders
WHERE created_at >= NOW() - INTERVAL '1 hour'
ORDER BY created_at;

# scripts/validation/compare.py
import pyodbc          # for SQL Server
import psycopg2        # for Postgres
import sys
from decimal import Decimal

def validate_table_counts(source_conn, target_conn, table_name):
    src_cur = source_conn.cursor()
    tgt_cur = target_conn.cursor()

    src_cur.execute(f"SELECT COUNT(*) FROM {table_name}")
    src_count = src_cur.fetchone()[0]

    tgt_cur.execute(f"SELECT COUNT(*) FROM {table_name}")
    tgt_count = tgt_cur.fetchone()[0]

    return {
        'check': 'row_count',
        'table': table_name,
        'source': src_count,
        'target': tgt_count,
        'status': 'PASS' if src_count == tgt_count else 'FAIL',
        'delta': tgt_count - src_count
    }

# ... similar for aggregates, distributions, etc.

def main():
    source = pyodbc.connect(SOURCE_CONN_STRING)
    target = psycopg2.connect(TARGET_CONN_STRING)

    results = []
    for table in TABLES_TO_VALIDATE:
        results.append(validate_table_counts(source, target, table))
        results.append(validate_aggregates(source, target, table))
        # ... etc

    report = generate_markdown_report(results)
    print(report)

    # Exit non-zero if any FAIL
    if any(r['status'] == 'FAIL' for r in results):
        sys.exit(1)

def spot_check_table(source_conn, target_conn, table_name, pk_column='id', sample_size=50):
    src_cur = source_conn.cursor()
    src_cur.execute(f"SELECT TOP {sample_size} * FROM {table_name} ORDER BY NEWID()")
    src_rows = src_cur.fetchall()
    src_columns = [desc[0] for desc in src_cur.description]

    differences = []
    for src_row in src_rows:
        pk_value = src_row[src_columns.index(pk_column)]
        tgt_cur = target_conn.cursor()
        tgt_cur.execute(f"SELECT * FROM {table_name} WHERE {pk_column} = %s", (pk_value,))
        tgt_row = tgt_cur.fetchone()

        if tgt_row is None:
            differences.append({'pk': pk_value, 'issue': 'Missing in target'})
            continue

        for col_idx, col_name in enumerate(src_columns):
            src_val = src_row[col_idx]
            tgt_val = tgt_row[col_idx]
            if not values_equal(src_val, tgt_val, col_name):
                differences.append({
                    'pk': pk_value,
                    'column': col_name,
                    'source': src_val,
                    'target': tgt_val
                })

    return differences

def values_equal(src, tgt, col_name):
    """Compare values with tolerance for known acceptable differences."""
    # Whitespace tolerance for legacy-stored text
    if isinstance(src, str) and isinstance(tgt, str):
        return src.strip() == tgt.strip()
    # Decimal precision tolerance
    if isinstance(src, Decimal) and isinstance(tgt, Decimal):
        return abs(src - tgt) < Decimal('0.001')
    # Skip migration-added timestamp fields
    if col_name in {'migrated_at', 'last_synced'}:
        return True
    return src == tgt

def drift_report():
    return {
        'timestamp': datetime.now(),
        'tables': {
            'customers': {
                'count_drift': 0,
                'sum_drift': Decimal('0.00'),
                'last_synced': '2024-01-15T12:00:00Z',
                'lag_seconds': 30
            },
            'orders': {
                'count_drift': 5,  # FAIL
                'sum_drift': Decimal('-127.50'),
                'last_synced': '2024-01-15T11:55:00Z',
                'lag_seconds': 330
            }
        }
    }

## Validation report — 2024-01-15 14:00 UTC

### customers
- ✅ Row count: 1,234,567 (match)
- ✅ Sum check: passed
- ✅ Distribution: matches across 5 status values
- ✅ FK integrity: 0 orphans
- ✅ Spot check: 50/50 rows match
- **Status: PASS**

### orders
- ❌ Row count: source=8,901,234, target=8,901,229, delta=-5
- ❌ Sum check: source=$45,678,123.50, target=$45,678,000.00, delta=-$123.50
- ⚠️ Distribution: matches except 'cancelled' (src=12,345, tgt=12,340)
- ✅ FK integrity: 0 orphans
- ❌ Spot check: 47/50 rows match. 3 differences:
  - id=42: column `discount` source=$10.00, target=$0.00
  - id=87: column `notes` source='Test order', target=NULL
  - id=156: column `status` source='cancelled', target='void'
- **Status: FAIL — 5 rows missing, $123.50 missing, 3 spot-check failures**

### Recommended actions
1. Investigate the 5 missing orders (likely the same orders that account for the $123.50)
2. Check status mapping: 'void' vs 'cancelled' — are these supposed to be the same enum value?
3. Re-run migration for affected rows OR backfill manually

## Validation results

[Per-table report with PASS/FAIL/WARN]

## Summary
- Tables checked: N
- Passed: N
- Failed: N
- Warnings: N

## Action items
[Specific, actionable next steps]

## Scripts generated
[Paths to validation scripts]

## Pairing with other skills

- **Data Migration Plan** template includes a section on validation; this skill generates the actual validation code
- **Migration Planner Skill** orchestrates this skill during the validation phase
- **Postgres MCP** + **Filesystem MCP** make this skill more useful — direct DB access for queries

## Tips

- Run validation continuously during dual-write phases. Don't wait until cutover to discover drift.
- Save baseline validation reports — comparing today's drift to last week's tells you if the system is degrading.
- For very large tables, sampling at the right scale matters. 50 rows out of 10M might miss systemic bugs; sample smarter (stratified by status, by date range).
- Pair with feature flags: validate per-tenant after each tenant migration.

## Limitations

- Cannot validate behavior, only data state. (Behavior parity tests handle behavior.)
- Cross-engine comparisons (SQL Server vs Postgres) require careful handling of type differences (decimal precision, datetime tz).
- For very large tables, full validation is expensive — generate sampled validation scripts for these.
- Cannot fix the drift, only report it. Remediation is a separate concern.