How do I validate data integrity after a cutover?

So you've executed your meticulously planned migration, celebrated the cutover, and now you're staring at the new system wondering if it's actually *correct*. The sales decks and migration guides go silent right at this point, don't they? All that talk about "seamless" and "zero-downtime" evaporates, leaving you with the cold, hard truth: you have two systems that *should* have the same data, and you have absolutely no proof that they do.

Everyone talks about pre-migration checks and the cutover itself, but the post-cutover validation is treated as an afterthought. It's the most critical phase. This is where you discover that your new "highly consistent" database interpreted that `TIMESTAMP` field differently, or that the event stream processor is dropping messages with certain payload sizes. You don't validate integrity by just checking row counts. A count is the laziest metric imaginable and tells you precisely nothing about the content.

Here's a pragmatic, cynical approach I've been forced to develop after being burned by several "trust us" migrations. You need a three-phase validation strategy, executed *after* cutover, while the old system is still a frozen artifact you can query.

**Phase 1: The Statistical Sampler**
Don't even think about a full-table scan comparison on day one. You'll melt something. Instead, write a script that randomly samples records by primary key or a known unique identifier. Compare critical fields, checksums, or entire JSON payloads. The key is to do this across a meaningful distribution of your data shards, partitions, or date ranges.

```python
import random
import hashlib
import psycopg2

def fetch_row_sample(conn, table, pk_column, sample_size=1000):
"""Fetch a random sample of primary keys and their row data."""
with conn.cursor() as cur:
cur.execute(f"SELECT {pk_column} FROM {table} TABLESAMPLE SYSTEM (0.1);")
all_pks = [row[0] for row in cur.fetchall()]
sampled_pks = random.sample(all_pks, min(sample_size, len(all_pks)))

rows = {}
for pk in sampled_pks:
cur.execute(f"SELECT * FROM {table} WHERE {pk_column} = %s", (pk,))
rows[pk] = cur.fetchone()
return rows

def hash_row(row):
"""Create a deterministic hash of a row, ignoring micro-timestamp differences."""
# Normalize data here (e.g., round timestamps, order JSON keys)
normalized = str(sorted(row.items()))
return hashlib.sha256(normalized.encode()).hexdigest()
```

**Phase 2: The Aggregate Truth Teller**
Sampling finds localized corruption. Aggregate comparisons find systemic bias. Run your critical business logic aggregates on both systems: sum of revenue, count of active users, average session length. Any discrepancy > your known tolerance (account for eventual consistency windows) is a red flag. This often catches issues like duplicate processing or dropped records in event pipelines.

**Phase 3: The Business Logic Smoke Test**
This is where you stop looking at data and start looking at outcomes. Run your key reporting queries, generate last week's financial report, rebuild your search index from the new source. Does the output match? Not just in number, but in shape? This is the only phase that truly matters to anyone outside the engineering team, and it's the one most often skipped in the rush to decommission the old, expensive system.

How long does this take? For a moderate dataset, the sampling and aggregates can be automated and run in a few hours. The business logic validation is a multi-day, sometimes multi-week, process of parallel runs and shadow traffic. Anyone who tells you they validated a major data migration in a single weekend is either lying or about to have a very interesting quarter.

-- cynical ops