feat: post-load data-quality gate for zero/null measure columns

Profile every MEASURE column after each Snowflake load (both dataset-first
and legacy paths) via INFORMATION_SCHEMA + per-table aggregates. Fail the
run loudly, naming the columns, when any measure is entirely zero/null
(recurring Yodeck/Triumph zero-measure defect). Mostly-zero and empty-table
cases surface as warnings in the completion panel and session_logs meta.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

chat_interface.py

@@ -3488,6 +3488,7 @@ LegitData will generate realistic, AI-powered data.
 
         # Clear and initialize live progress for Snowflake deployment
         self.live_progress_log = ["=" * 60, "SNOWFLAKE DEPLOYMENT STARTING", "=" * 60, ""]
+        self._dq_warnings = []  # reset data-quality gate warnings for this run
 
         def log_progress(msg):
             """Log to live progress tab only — not pipeline status"""
@@ -3717,13 +3718,67 @@ LegitData will generate realistic, AI-powered data.
             
             progress += f"\n[OK] Data populated"
             log_progress(f"[OK] {pop_message}")
-            
+
+            # Data-quality gate: after every load, profile the MEASURE columns
+            # that were actually written to Snowflake and fail loudly if any is
+            # entirely zero/null (recurring zero-measure defect — derived
+            # measures like TOTAL_*_USD loading as all zeros → blank tiles).
+            from demoprep_app.integrations.snowflake import (
+                ZeroMeasureError,
+                run_measure_quality_gate,
+            )
+            progress += f"\n\n**Verifying data quality (measure columns)...**"
+            yield progress
+            log_progress("")
+            log_progress("Running data-quality gate on measure columns...")
+            try:
+                dq_profile = run_measure_quality_gate(
+                    deployer.connection, schema_name, progress_callback=log_progress
+                )
+            except ZeroMeasureError as e:
+                self._last_population_error = str(e)
+                self._last_schema_name = schema_name
+                _deploy_meta = {
+                    "schema_name": schema_name,
+                    "data_quality": {
+                        "failures": e.failures,
+                        "warnings": e.profile.get("warnings", []),
+                    },
+                }
+                log_progress(f"[ERROR] {e}")
+                log_progress(f"  Schema {schema_name} left in place for inspection.")
+                if _slog:
+                    _slog.log("deploy", "data quality gate failed", error=str(e))
+                raise Exception(str(e))
+
+            dq_warnings = dq_profile.get("warnings", [])
+            self._dq_warnings = dq_warnings
+            progress += (
+                f"\n[OK] Data quality gate passed "
+                f"({dq_profile['measure_columns_checked']} measure columns "
+                f"across {dq_profile['tables_checked']} tables)"
+            )
+            for _dq_w in dq_warnings:
+                progress += f"\n[WARN] {_dq_w}"
+            if _slog:
+                _slog.log_verbose(
+                    "deploy", "data quality gate passed",
+                    measure_columns_checked=dq_profile["measure_columns_checked"],
+                    tables_checked=dq_profile["tables_checked"],
+                    dq_warnings=dq_warnings,
+                )
+
             self._deployed_schema_name = schema_name
             _deploy_meta = {
                 "schema_name": schema_name,
                 "generation_mode": "dataset_first" if self._dataset_first_bundle else "legacy",
                 "table_names": [table.name for table in self._dataset_first_bundle.tables] if self._dataset_first_bundle else [],
                 "row_counts": {table.name: len(table.rows) for table in self._dataset_first_bundle.tables} if self._dataset_first_bundle else {},
+                "data_quality": {
+                    "measure_columns_checked": dq_profile["measure_columns_checked"],
+                    "tables_checked": dq_profile["tables_checked"],
+                    "warnings": dq_warnings,
+                },
             }
             
             log_progress("")
@@ -4468,7 +4523,13 @@ The liveboard GUID couldn't be retrieved — it may still have been created in T
 - Type **'retry liveboard'** to try building it again
 
 Type **'done'** to finish."""
-                
+
+                # Surface data-quality warnings from the load gate in the completion panel
+                _dq_warnings = getattr(self, '_dq_warnings', None)
+                if _dq_warnings:
+                    response += "\n\n---\n\n**⚠️ Data quality warnings:**\n" + \
+                        "\n".join(f"- {w}" for w in _dq_warnings)
+
                 yield {'response': response, 'stage': final_stage}
             else:
                 errors = results.get('errors', ['Unknown error'])

demoprep_app/integrations/snowflake/__init__.py

@@ -1,6 +1,10 @@
 """Snowflake integration helpers."""
 
 from demoprep_app.integrations.snowflake.dataset_writer import populate_dataset_bundle
+from demoprep_app.integrations.snowflake.quality_gate import (
+    ZeroMeasureError,
+    run_measure_quality_gate,
+)
 
-__all__ = ["populate_dataset_bundle"]
+__all__ = ["populate_dataset_bundle", "run_measure_quality_gate", "ZeroMeasureError"]
 

demoprep_app/integrations/snowflake/quality_gate.py

@@ -0,0 +1,170 @@
+"""Post-load data-quality gate for generated demo schemas.
+
+Guards against the recurring zero-measure defect: fact tables where derived
+measure columns (e.g. TOTAL_LOAD_REVENUE_USD, BROKER_MARGIN_USD) load as 100%
+zero/null while their input columns are populated, which renders liveboard
+tiles blank. Runs immediately after data population, against what was actually
+loaded into Snowflake, so it covers both the dataset-first and legacy
+LegitData paths.
+
+The gate profiles every MEASURE column (numeric columns that are not join
+keys, mirroring ThoughtSpotDeployer._determine_column_type) and raises
+ZeroMeasureError when any measure is entirely zero/null. No silent fallback:
+a failed gate fails the run with the offending columns named.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Callable, Optional
+
+# Snowflake normalizes numeric types, but accept the common aliases too.
+_NUMERIC_DATA_TYPES = {
+    "NUMBER", "DECIMAL", "NUMERIC", "INT", "INTEGER", "BIGINT", "SMALLINT",
+    "TINYINT", "BYTEINT", "FLOAT", "FLOAT4", "FLOAT8", "DOUBLE",
+    "DOUBLE PRECISION", "REAL",
+}
+
+# Share of non-zero values below which a populated measure column is suspicious
+# (a partial variant of the all-zero defect) but not an outright failure.
+MOSTLY_ZERO_WARN_THRESHOLD = 0.01
+
+
+class ZeroMeasureError(Exception):
+    """One or more measure columns loaded entirely zero/null."""
+
+    def __init__(self, failures: list[dict], profile: dict):
+        self.failures = failures
+        self.profile = profile
+        details = "; ".join(
+            f"{f['table']}.{f['column']} ({f['row_count']} rows, all zero/null)"
+            for f in failures
+        )
+        super().__init__(
+            f"Data quality gate failed — {len(failures)} measure column(s) are "
+            f"entirely zero/null: {details}. The data generator did not compute "
+            f"these derived measures; the run must be regenerated."
+        )
+
+
+def is_measure_column(column_name: str, data_type: str) -> bool:
+    """Mirror ThoughtSpotDeployer._determine_column_type's MEASURE test."""
+    base_type = data_type.upper().split("(")[0].strip()
+    if base_type not in _NUMERIC_DATA_TYPES:
+        return False
+    col_upper = column_name.upper()
+    # Join keys are ATTRIBUTEs in the model, not measures.
+    if col_upper.endswith("ID") or col_upper.endswith("KEY") or col_upper.endswith("_CODE"):
+        return False
+    return True
+
+
+def _quote_ident(identifier: str) -> str:
+    escaped = identifier.replace('"', '""')
+    return f'"{escaped}"'
+
+
+def run_measure_quality_gate(
+    connection: Any,
+    schema_name: str,
+    progress_callback: Optional[Callable[[str], None]] = None,
+) -> dict:
+    """Profile every measure column in the schema; fail loudly on all-zero ones.
+
+    Returns a profile dict:
+        {
+            "tables_checked": int,
+            "measure_columns_checked": int,
+            "failures": [],  # empty on success (non-empty raises instead)
+            "warnings": ["FACT_X.COL is 99.8% zero/null (1/500 non-zero)", ...],
+            "columns": {"TABLE.COLUMN": {"row_count": n, "non_zero": m}, ...},
+        }
+
+    Raises ZeroMeasureError when any measure column in a populated table is
+    entirely zero/null. Profiling errors (bad schema name, connection loss)
+    propagate as-is — the gate never passes silently.
+    """
+    def log(msg: str) -> None:
+        if progress_callback:
+            progress_callback(msg)
+
+    cursor = connection.cursor()
+    try:
+        cursor.execute(
+            """
+            SELECT TABLE_NAME, COLUMN_NAME, DATA_TYPE
+            FROM INFORMATION_SCHEMA.COLUMNS
+            WHERE TABLE_SCHEMA = %s
+            ORDER BY TABLE_NAME, ORDINAL_POSITION
+            """,
+            (schema_name,),
+        )
+        rows = cursor.fetchall()
+        if not rows:
+            raise RuntimeError(
+                f"Data quality gate could not find any columns for schema "
+                f"'{schema_name}' in INFORMATION_SCHEMA — cannot verify the load."
+            )
+
+        measures_by_table: dict[str, list[str]] = {}
+        for table_name, column_name, data_type in rows:
+            if is_measure_column(column_name, data_type or ""):
+                measures_by_table.setdefault(table_name, []).append(column_name)
+
+        profile: dict = {
+            "tables_checked": 0,
+            "measure_columns_checked": 0,
+            "failures": [],
+            "warnings": [],
+            "columns": {},
+        }
+
+        for table_name, columns in sorted(measures_by_table.items()):
+            select_parts = ["COUNT(*)"]
+            for col in columns:
+                q = _quote_ident(col)
+                select_parts.append(f"SUM(IFF({q} IS NOT NULL AND {q} <> 0, 1, 0))")
+            cursor.execute(
+                f"SELECT {', '.join(select_parts)} "
+                f"FROM {_quote_ident(schema_name)}.{_quote_ident(table_name)}"
+            )
+            result = cursor.fetchone()
+            row_count = int(result[0] or 0)
+
+            profile["tables_checked"] += 1
+            if row_count == 0:
+                profile["warnings"].append(
+                    f"{table_name} has 0 rows — measure columns could not be verified"
+                )
+                continue
+
+            for idx, col in enumerate(columns):
+                non_zero = int(result[idx + 1] or 0)
+                profile["measure_columns_checked"] += 1
+                profile["columns"][f"{table_name}.{col}"] = {
+                    "row_count": row_count,
+                    "non_zero": non_zero,
+                }
+                if non_zero == 0:
+                    profile["failures"].append(
+                        {"table": table_name, "column": col, "row_count": row_count}
+                    )
+                elif non_zero / row_count < MOSTLY_ZERO_WARN_THRESHOLD:
+                    pct_zero = 100.0 * (1 - non_zero / row_count)
+                    profile["warnings"].append(
+                        f"{table_name}.{col} is {pct_zero:.1f}% zero/null "
+                        f"({non_zero}/{row_count} non-zero)"
+                    )
+
+        log(
+            f"Data quality gate: profiled {profile['measure_columns_checked']} measure "
+            f"column(s) across {profile['tables_checked']} table(s)"
+        )
+        for warning in profile["warnings"]:
+            log(f"[WARN] Data quality: {warning}")
+
+        if profile["failures"]:
+            raise ZeroMeasureError(profile["failures"], profile)
+
+        return profile
+    finally:
+        cursor.close()

sprint_2026_04.md

@@ -221,6 +221,7 @@ should tell. KPI targets, growth trends, and anomaly patterns live in the matrix
 - [x] **AI Viz Titles** — `_humanize_viz_titles()` in `liveboard_creator.py`; one LLM call renames all raw TS column-name titles to business-readable labels; runs as Step 6.5 in `enhance_mcp_liveboard()` before TML re-import ✅
 - [x] **KPI conversion fix** — Step 3.5 no longer promotes "by X" dimensional breakdowns to KPIs (was putting "Averagesellingprice by Category Weekly" in the Key Metrics group) ✅
 - [x] **Chart variety** — multi-dim breakdowns ("by X and Y") → STACKED_COLUMN; single-dim categorical breakdowns now catches LINE charts (MCP often generates LINE for categoricals, blocking donut conversion) ✅
+- [x] **Post-load measure data-quality gate** (Aug 25) — `demoprep_app/integrations/snowflake/quality_gate.py`; runs in `run_deployment_streaming()` right after population (both dataset-first and legacy paths); profiles every MEASURE column via INFORMATION_SCHEMA + per-table aggregates; raises `ZeroMeasureError` naming the columns when any measure loads entirely zero/null (Yodeck/Triumph zero-measure defect); mostly-zero (<1% non-zero) and empty tables → warnings surfaced in completion panel + `_deploy_meta.data_quality` in session_logs for grading; unit tests in `tests/test_quality_gate.py` ✅
 
 ### Shipped at end of Sprint 3 / mini sprint (Apr 28-29)
 

tests/test_quality_gate.py

@@ -0,0 +1,142 @@
+"""Unit tests for the post-load measure data-quality gate.
+
+Run with: pytest tests/test_quality_gate.py -v
+"""
+
+import pytest
+
+from demoprep_app.integrations.snowflake.quality_gate import (
+    ZeroMeasureError,
+    is_measure_column,
+    run_measure_quality_gate,
+)
+
+
+class FakeCursor:
+    """Answers the INFORMATION_SCHEMA query, then per-table aggregate queries."""
+
+    def __init__(self, columns_rows, table_aggregates):
+        # columns_rows: list of (table_name, column_name, data_type)
+        # table_aggregates: {table_name: (row_count, nonzero_col1, nonzero_col2, ...)}
+        self._columns_rows = columns_rows
+        self._table_aggregates = table_aggregates
+        self._last_result = None
+        self.executed = []
+
+    def execute(self, sql, params=None):
+        self.executed.append(sql)
+        if "INFORMATION_SCHEMA.COLUMNS" in sql:
+            self._last_result = self._columns_rows
+        else:
+            table = next(t for t in self._table_aggregates if f'"{t}"' in sql)
+            self._last_result = self._table_aggregates[table]
+
+    def fetchall(self):
+        return self._last_result
+
+    def fetchone(self):
+        return self._last_result
+
+    def close(self):
+        pass
+
+
+class FakeConnection:
+    def __init__(self, cursor):
+        self._cursor = cursor
+
+    def cursor(self):
+        return self._cursor
+
+
+def test_is_measure_column_heuristic():
+    # Numeric non-key columns are measures
+    assert is_measure_column("TOTAL_LOAD_REVENUE_USD", "NUMBER(12,2)")
+    assert is_measure_column("BROKER_MARGIN_PCT", "FLOAT")
+    # Join keys and codes are not measures
+    assert not is_measure_column("LOAD_ID", "NUMBER(38,0)")
+    assert not is_measure_column("LANE_KEY", "NUMBER(38,0)")
+    assert not is_measure_column("CARRIER_CODE", "NUMBER(38,0)")
+    # Non-numeric columns are not measures
+    assert not is_measure_column("ORIGIN_CITY", "VARCHAR(100)")
+    assert not is_measure_column("PICKUP_DATE", "DATE")
+
+
+def test_gate_fails_loudly_on_all_zero_measure():
+    cursor = FakeCursor(
+        columns_rows=[
+            ("FACT_LOAD_TRANSACTION", "LOAD_ID", "NUMBER"),
+            ("FACT_LOAD_TRANSACTION", "LOADS_ACCEPTED", "NUMBER"),
+            ("FACT_LOAD_TRANSACTION", "TOTAL_LOAD_REVENUE_USD", "NUMBER"),
+            ("FACT_LOAD_TRANSACTION", "BROKER_MARGIN_USD", "NUMBER"),
+        ],
+        # 500 rows: LOADS_ACCEPTED populated, both derived measures all zero
+        table_aggregates={"FACT_LOAD_TRANSACTION": (500, 500, 0, 0)},
+    )
+
+    with pytest.raises(ZeroMeasureError) as exc_info:
+        run_measure_quality_gate(FakeConnection(cursor), "TRI_TEST_sch")
+
+    err = exc_info.value
+    assert len(err.failures) == 2
+    failed_columns = {f["column"] for f in err.failures}
+    assert failed_columns == {"TOTAL_LOAD_REVENUE_USD", "BROKER_MARGIN_USD"}
+    # Error message must name the offending columns (fail loudly, no guessing)
+    assert "FACT_LOAD_TRANSACTION.TOTAL_LOAD_REVENUE_USD" in str(err)
+    assert "FACT_LOAD_TRANSACTION.BROKER_MARGIN_USD" in str(err)
+    # ID column must not be profiled as a measure
+    assert not any(f["column"] == "LOAD_ID" for f in err.failures)
+
+
+def test_gate_passes_with_populated_measures():
+    cursor = FakeCursor(
+        columns_rows=[
+            ("FACT_SALES", "SALE_ID", "NUMBER"),
+            ("FACT_SALES", "REVENUE_USD", "NUMBER"),
+            ("DIM_PRODUCT", "PRODUCT_ID", "NUMBER"),
+            ("DIM_PRODUCT", "UNIT_PRICE", "NUMBER"),
+        ],
+        table_aggregates={
+            "FACT_SALES": (500, 498),
+            "DIM_PRODUCT": (50, 50),
+        },
+    )
+
+    profile = run_measure_quality_gate(FakeConnection(cursor), "SALES_sch")
+    assert profile["failures"] == []
+    assert profile["tables_checked"] == 2
+    assert profile["measure_columns_checked"] == 2
+    assert profile["columns"]["FACT_SALES.REVENUE_USD"] == {
+        "row_count": 500,
+        "non_zero": 498,
+    }
+
+
+def test_gate_warns_on_mostly_zero_measure():
+    cursor = FakeCursor(
+        columns_rows=[("FACT_SALES", "DISCOUNT_USD", "NUMBER")],
+        # 1 non-zero value in 500 rows (0.2%) — suspicious but not a failure
+        table_aggregates={"FACT_SALES": (500, 1)},
+    )
+
+    profile = run_measure_quality_gate(FakeConnection(cursor), "SALES_sch")
+    assert profile["failures"] == []
+    assert len(profile["warnings"]) == 1
+    assert "FACT_SALES.DISCOUNT_USD" in profile["warnings"][0]
+
+
+def test_gate_warns_on_empty_table():
+    cursor = FakeCursor(
+        columns_rows=[("FACT_SALES", "REVENUE_USD", "NUMBER")],
+        table_aggregates={"FACT_SALES": (0, 0)},
+    )
+
+    profile = run_measure_quality_gate(FakeConnection(cursor), "SALES_sch")
+    assert profile["failures"] == []
+    assert any("0 rows" in w for w in profile["warnings"])
+
+
+def test_gate_errors_when_schema_has_no_columns():
+    cursor = FakeCursor(columns_rows=[], table_aggregates={})
+    with pytest.raises(RuntimeError, match="could not find any columns"):
+        run_measure_quality_gate(FakeConnection(cursor), "MISSING_sch")