data_quality_gate.py

"""Semantic data-quality gate — run between data load and liveboard creation.

Motivation (Yodeck, Aug 20 2026): the generator wrote every derived measure
(ARR, Net New MRR, NRR, collection/margin rates) as all-zero, and the liveboard
charted exactly those columns — 6 of 10 vizzes rendered nothing or $0. Nothing
in the pipeline checked that charted data meant anything: every gate was a
plumbing gate (DDL compiled, rows loaded, TML imported).

This module is the semantic gate:
  * scan_dead_measures(database, schema): one aggregate query per table finds
    numeric columns that are entirely NULL/zero ("dead") or constant.
  * filter_model_columns(model_columns, dead_names): drops model MEASURE
    columns whose physical column is dead, so question/KPI/viz generation can
    only reach measures with real data.

Fail-open by design: any error here must never break a build — callers wrap in
try/except and proceed unfiltered with a warning.
"""

from __future__ import annotations

import re
from typing import Dict, List, Tuple

# Numeric columns that are join/surrogate keys, not measures — never scanned.
_KEY_SUFFIXES = ("_KEY", "_ID")

# Snowflake numeric types eligible for the dead-measure scan.
_NUMERIC_TYPES = ("NUMBER", "FLOAT", "FIXED", "REAL", "DECIMAL", "NUMERIC", "INT")


def _normalize(name: str) -> str:
    """Match model display names to physical columns: 'Net New Mrr' == NET_NEW_MRR."""
    return re.sub(r"[^a-z0-9]", "", (name or "").lower())


def scan_dead_measures(database: str, schema: str, log=None) -> Dict:
    """Scan every base table in the schema for dead numeric measure columns.

    Returns {
      'dead':     {table: [column, ...]},   # all rows NULL or zero
      'constant': {table: [column, ...]},   # single repeated non-zero value
      'dead_names': set of normalized dead column names (for model filtering),
      'warnings': [str, ...],
    }
    Opens its own Snowflake connection (keypair auth via snowflake_auth).
    """
    def _log(msg: str) -> None:
        if log:
            log(msg)

    from snowflake_auth import get_snowflake_connection

    out = {"dead": {}, "constant": {}, "dead_names": set(), "warnings": []}
    conn = get_snowflake_connection()
    try:
        cur = conn.cursor()
        cur.execute(
            f"""
            select c.table_name, c.column_name
            from {database}.information_schema.columns c
            join {database}.information_schema.tables t
              on t.table_schema = c.table_schema and t.table_name = c.table_name
            where c.table_schema = %s
              and t.table_type = 'BASE TABLE'
              and c.data_type in ({','.join("'" + t + "'" for t in _NUMERIC_TYPES)})
            order by c.table_name, c.ordinal_position
            """,
            (schema,),
        )
        by_table: Dict[str, List[str]] = {}
        for table, column in cur.fetchall():
            if column.upper().endswith(_KEY_SUFFIXES):
                continue
            by_table.setdefault(table, []).append(column)

        for table, cols in by_table.items():
            # One aggregate pass per table: [nonzero-count, min, max] per column.
            selects = []
            for c in cols:
                selects.append(f'count_if("{c}" is not null and "{c}" <> 0)')
                selects.append(f'min("{c}")')
                selects.append(f'max("{c}")')
            cur.execute(
                f'select count(*), {", ".join(selects)} from {database}."{schema}"."{table}"'
            )
            row = cur.fetchone()
            total = row[0]
            if not total:
                continue
            for i, c in enumerate(cols):
                nonzero, cmin, cmax = row[1 + i * 3], row[2 + i * 3], row[3 + i * 3]
                if nonzero == 0:
                    out["dead"].setdefault(table, []).append(c)
                    out["dead_names"].add(_normalize(c))
                elif total > 1 and cmin == cmax:
                    out["constant"].setdefault(table, []).append(c)

        for table, cols in out["dead"].items():
            msg = f"DATA GATE: {table} has all-zero/all-null measures: {', '.join(cols)}"
            out["warnings"].append(msg)
            _log(f"   [GATE] {msg}")
        for table, cols in out["constant"].items():
            msg = f"DATA GATE: {table} has constant-value measures: {', '.join(cols)}"
            out["warnings"].append(msg)
            _log(f"   [GATE] {msg}")
        if not out["dead"] and not out["constant"]:
            _log("   [GATE] Data gate clean: no dead or constant measures")
    finally:
        conn.close()
    return out


def filter_model_columns(model_columns: List[Dict], dead_names) -> Tuple[List[Dict], List[str]]:
    """Split model columns into (kept, excluded_names). Only MEASURE-typed
    columns are eligible for exclusion — attributes/dates always pass."""
    kept, excluded = [], []
    for col in model_columns or []:
        name = col.get("name") or ""
        ctype = (col.get("type") or "").upper()
        if ctype == "MEASURE" and _normalize(name) in dead_names:
            excluded.append(name)
        else:
            kept.append(col)
    return kept, excluded