Data Quality and System Integration Challenges Questions

import re
from typing import Dict, Tuple, List

EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")

def validate_record(rec: Dict) -> Tuple[bool, List[str]]:
    """
    Validate a single user record.
    Returns (is_valid, list_of_errors).
    """
    errors = []

    # Required fields
    if "user_id" not in rec or rec["user_id"] in (None, ""):
        errors.append("missing user_id")
    if "email" not in rec or rec["email"] in (None, ""):
        errors.append("missing email")
    else:
        # Email format
        if not EMAIL_RE.match(rec["email"]):
            errors.append("invalid email format")

    # Optional age
    if "age" in rec and rec["age"] not in (None, ""):
        try:
            age = int(rec["age"])
            if not (0 <= age <= 120):
                errors.append("age out of range")
        except (ValueError, TypeError):
            errors.append("age not an integer")

    return (len(errors) == 0, errors)

-- Example: BigQuery / Postgres-like SQL with MD5 and STRING_AGG
SELECT
  partition_key,
  COUNT(*)        AS row_count,
  MIN(updated_at) AS min_ts,
  MAX(updated_at) AS max_ts,
  MD5(STRING_AGG(row_hash, '' ORDER BY pk)) AS partition_checksum
FROM (
  SELECT
    pk,
    MD5(CONCAT_WS('||', COALESCE(col1,''), COALESCE(col2,''), COALESCE(col3,''), COALESCE(CAST(updated_at AS TEXT),''))) AS row_hash
  FROM dataset.table
  WHERE partition_date = @date
) t
GROUP BY partition_key;

-- Assume checksums_source and checksums_warehouse are small tables
SELECT s.partition_key
FROM checksums_source s
FULL OUTER JOIN checksums_warehouse w USING (partition_key)
WHERE s.partition_checksum IS DISTINCT FROM w.partition_checksum
   OR s.row_count IS DISTINCT FROM w.row_count

SELECT s.pk, s.row_hash AS src_hash, w.row_hash AS wh_hash
FROM src_partition s
FULL OUTER JOIN wh_partition w USING (pk)
WHERE s.row_hash IS DISTINCT FROM w.row_hash;