Experimentation System: Data Foundation

Auto-generate Pydantic models from Unity Catalog and populate real metric baselines.

Introduction

30 Second Summary

Experiments need real numbers to produce trustworthy results. If your metrics still carry placeholder baselines, every sample size calculation you run is guesswork dressed up as math.

In this project, you will transform your Marketing Experimentation System from demo mode to production-ready by generating schema contracts from your live tables and populating every metric with real baseline values from your CRM and marketing data.

What You'll Build

You'll run the full pipeline and watch your 17 Delta tables transform into validated Python models, see real conversion rates flow into your metrics dictionary segmented by account tier, and trigger a CI gate that catches schema drift the moment it happens.

By the end of this project, you'll have:

  • A complete Pydantic model library auto-generated from your live Unity Catalog schema, with B2B-specific validation constraints that reject invalid data at the code level.
  • Every metric in your metrics_dictionary populated with real baseline values, variance estimates, and accrual rates segmented by Enterprise, Mid-Market, and SMB accounts.
  • A validation suite that executes all metric SQL definitions against live source tables and confirms they return expected results before any experiment runs.
  • Secret Mission: Build an automated baseline drift detector that alerts when any metric shifts more than 20% from its stored value.

Are there any prerequisites?

This project builds on an existing MES deployment. You need your 17-table schema already deployed in Unity Catalog, a Databricks workspace with SQL warehouse access, and a Python environment with databricks-sdk installed.

You should also have your statistical engine and analysis runbooks in place from the earlier MES projects.

Before We Start

Before any code changes or queries, let's lock in what you're about to build and why it matters. This step is your commitment to the specific outcome this production hardening delivers for your experimentation program.

Set Up Pydantic and Schema Introspection

Your 17 Delta tables have DDL definitions in Unity Catalog, but nothing in your Python code enforces that contract. If a column gets renamed or a type changes, your app code silently produces garbage downstream.

In this step, you will install Pydantic and build the introspection query that reads your live schema from Unity Catalog's information_schema. By the end, you will have confirmed all 17 tables are reflected with their full column definitions.

In this step, get ready to:
  • Install Pydantic v2 and confirm your environment is ready.
  • Write the information_schema introspection query in a new script.
  • Run the query and confirm all 17 MES tables are returned with their column definitions.
Install Pydantic v2 and configure your environment

Your project needs Pydantic for type-safe data models and the Databricks SDK for querying Unity Catalog. The SDK is already in your environment, but Pydantic needs to be installed fresh.

  • Install Pydantic by running this command in your terminal:
pip install pydantic
  • Confirm the installation succeeded by checking the version:
python -c "import pydantic; print(pydantic.__version__)"

You should see a version starting with 2. (such as 2.13.4).

Seeing a version starting with 1.x?

If you see a version like 1.10.x, you have Pydantic v1 installed. Uninstall it first with pip uninstall pydantic, then reinstall.

help me upgrade from Pydantic v1 to v2

  • Set your SQL warehouse ID as an environment variable by running:
export DATABRICKS_WAREHOUSE_ID=your-warehouse-id-here

Where do I find my warehouse ID?

In your Databricks workspace, navigate to SQL Warehouses in the left sidebar. Click on your warehouse name, then look at the URL or the Connection Details tab for the ID string.

  • Confirm the Databricks SDK is available by running:
python -c "from databricks.sdk import WorkspaceClient; print('SDK ready')"

You should see SDK ready printed with no import errors.

Seeing a ModuleNotFoundError?

If the SDK import fails, install it with pip install databricks-sdk. Make sure you are using the same Python environment where your MES project runs.

help me fix my databricks-sdk import error

Write the information_schema introspection query

Unity Catalog exposes an information_schema that describes every table, column, and data type in your catalog. You will query it to pull the full column definitions for all 17 MES tables, then use that metadata to generate Pydantic models in the next step.

  • Create a new file called schemas_generator.py in your MES project folder.
  • Add the imports and type-mapping constants by pasting this code:
from databricks.sdk import WorkspaceClient
import os

CATALOG = "marketing_experimentation"
SCHEMA = "mes_dev"

# Maps Databricks SQL types to Python type strings
TYPE_MAP = {
    "STRING": "str",
    "INT": "int",
    "BIGINT": "int",
    "SMALLINT": "int",
    "TINYINT": "int",
    "DOUBLE": "float",
    "FLOAT": "float",
    "DECIMAL": "Decimal",
    "BOOLEAN": "bool",
    "DATE": "date",
    "TIMESTAMP": "datetime",
    "ARRAY": "list",
    "MAP": "dict",
    "STRUCT": "dict",
}

What does this code do?

CATALOG and SCHEMA pin the query to your specific Unity Catalog location. TYPE_MAP defines how each Databricks SQL type will translate to a Python type when you generate Pydantic models in Step 2.

  • Save schemas_generator.py.
  • Verify the constants load cleanly by running:
python -c "from schemas_generator import TYPE_MAP, CATALOG; print(f'{len(TYPE_MAP)} type mappings loaded for {CATALOG}')"

You should see 14 type mappings loaded for marketing_experimentation.

  • Next, add the B2B validation rules and enum definitions below TYPE_MAP:
# B2B-specific field constraints for generated models
FIELD_VALIDATORS = {
    "mde_pct": {"ge": 0.05, "le": 0.50},
    "alpha": {"ge": 0.01, "le": 0.10},
    "power": {"ge": 0.70, "le": 0.99},
    "split_ratio": {"ge": 0.10, "le": 0.90},
}

# Lifecycle enums that status fields validate against
ENUM_DEFINITIONS = {
    "ExperimentStatus": ["proposed", "designed", "approved", "running", "analysed", "decided", "closed"],
    "Method": ["paired_geo", "sequential_ab", "ab_test", "abn_test", "did_holdout", "geo_synthetic", "switchback", "iroas", "matched_account", "pre_post_counterfactual"],
    "AssignmentUnit": ["account", "geo", "contact", "campaign_cohort"],
    "DecisionOutcome": ["ship", "scale", "kill", "iterate", "inconclusive", "scale_pending_confirmation"],
    "MetricType": ["proportion", "continuous", "ratio", "count"],
    "ReadoutType": ["interim", "final", "post_hoc", "confirmation"],
    "ImplementationStatus": ["not_applicable", "pending", "implemented", "declined"],
    "ConfirmationStatus": ["pending", "confirmed", "not_confirmed", "not_applicable"],
}

What do these constants control?

FIELD_VALIDATORS defines hard boundaries for your B2B experiment parameters. For example, mde_pct is constrained to 0.05-0.50 because detecting effects below 5% requires impractical sample sizes in B2B, and targeting above 50% means you do not need a test.

ENUM_DEFINITIONS lists the valid lifecycle values for each status field. These match the enums you defined in your original DDL.

  • Now add the core introspection function that queries Unity Catalog. Paste this below ENUM_DEFINITIONS:
def get_table_schemas(warehouse_id: str) -> dict:
    """Query information_schema for all MES table columns."""
    w = WorkspaceClient()
    query = f"""
    SELECT table_name, column_name, data_type, is_nullable
    FROM {CATALOG}.information_schema.columns
    WHERE table_catalog = '{CATALOG}'
      AND table_schema = '{SCHEMA}'
    ORDER BY table_name, ordinal_position
    """
    result = w.statement_execution.execute_statement(
        warehouse_id=warehouse_id,
        statement=query
    )
    # Group columns by table name
    tables = {}
    for row in result.result.data_array:
        table_name = row[0]
        if table_name not in tables:
            tables[table_name] = []
        tables[table_name].append({
            "column_name": row[1],
            "data_type": row[2],
            "is_nullable": row[3] == "YES",
        })
    return tables

What does get_table_schemas do?

This function queries information_schema.columns which Unity Catalog exposes for every catalog. The WHERE clause filters to only your mes_dev schema, preventing query timeouts from scanning the entire catalog.

The result is a dictionary keyed by table name, where each value is a list of column definitions (name, type, nullability). Unity Catalog's information_schema is privilege-aware, so it automatically filters to only objects you can access.

  • Save schemas_generator.py.
Run the introspection query and confirm your 17 tables

The script has everything needed to pull your live schema. Before you run it, think about what you expect to see.

Do you expect all 17 tables to appear, or might some be missing? What could cause a table to not show up in the results?

  • Run the introspection query against your live schema:
python -c "
from schemas_generator import get_table_schemas
import os
tables = get_table_schemas(os.environ['DATABRICKS_WAREHOUSE_ID'])
print(f'Found {len(tables)} tables')
for name in sorted(tables.keys()):
    print(f'  {name}: {len(tables[name])} columns')
"

You should see exactly 17 tables listed, each with their full column count. Every table from your MES schema (experiments, metrics_dictionary, design_decisions, readouts, and the rest) appears with its complete column definitions.

Seeing fewer than 17 tables?

Confirm your DATABRICKS_WAREHOUSE_ID environment variable is set correctly. Run echo $DATABRICKS_WAREHOUSE_ID to verify it is not empty.

If you see 0 tables, check that your Databricks authentication is configured (the SDK reads from ~/.databrickscfg or environment variables).

help me debug why my information_schema query returns fewer tables than expected

  • Now add the two utility functions that the model generator will need. Paste these below get_table_schemas:
def map_sql_type(sql_type: str) -> str:
    """Map a Databricks SQL type to a Python type string."""
    base_type = sql_type.split("(")[0].upper().strip()
    if base_type.startswith("ARRAY"):
        return "list"
    if base_type.startswith("MAP"):
        return "dict"
    if base_type.startswith("STRUCT"):
        return "dict"
    if base_type.startswith("DECIMAL"):
        return "Decimal"
    return TYPE_MAP.get(base_type, "str")


def table_name_to_class_name(table_name: str) -> str:
    """Convert snake_case table name to PascalCase class name."""
    return "".join(word.capitalize() for word in table_name.split("_"))

What do these utility functions do?

map_sql_type handles the edge cases in Databricks type names. Types like DECIMAL(10,2) or ARRAY<STRING> include parameters, so the function strips those before looking up the base type in TYPE_MAP.

table_name_to_class_name converts snake_case table names (like design_decisions) into PascalCase class names (DesignDecisions) for the Pydantic models.

  • Save schemas_generator.py.
  • Verify the utilities work by running a quick test:
python -c "
from schemas_generator import map_sql_type, table_name_to_class_name
print(map_sql_type('DECIMAL(10,2)'))  # should print: Decimal
print(map_sql_type('ARRAY<STRING>'))  # should print: list
print(table_name_to_class_name('design_decisions'))  # should print: DesignDecisions
"

You should see Decimal, list, and DesignDecisions printed on separate lines.

Seeing an ImportError?

Make sure you saved schemas_generator.py after adding the utility functions. Also confirm you are running the command from the same directory where the file is saved.

help me fix import errors when testing my schemas_generator utility functions

✔️ Awesome, I've got everything!

Great. Confirm you have saved schemas_generator.py with all the code from this step.

ⓧ I'd like to double check the full code

Your schemas_generator.py should contain these sections in order. Scroll up and confirm each is present:

  • Imports (WorkspaceClient, os) plus CATALOG and SCHEMA constants.
  • TYPE_MAP dictionary with 14 SQL-to-Python type entries.
  • FIELD_VALIDATORS dictionary with 4 constrained fields (mde_pct, alpha, power, split_ratio).
  • ENUM_DEFINITIONS dictionary with 8 enum types.
  • get_table_schemas function that queries information_schema.columns.
  • map_sql_type function that handles parameterized types.
  • table_name_to_class_name function that converts to PascalCase.

Your introspection script can now pull the full schema of every MES table from Unity Catalog. Next, you will use this metadata to auto-generate Pydantic models that enforce the schema as a code-level contract.

Generate Pydantic Models from Live Schema

You confirmed that all 17 tables are visible through information_schema in the last step. Now the challenge is turning that raw metadata into enforceable contracts in your Python code.

Manually writing Pydantic models for 17 tables means they drift from reality within weeks. Auto-generating from the live DDL means the models are always the source of truth, and schema changes surface immediately as validation failures.

In this step, get ready to:
  • Build a model generator function that emits one Pydantic BaseModel per table with correct type mappings and B2B validation rules.
  • Assemble the full file generator that produces enum classes and model classes from your live schema.
  • Run the generator and verify your models import cleanly with field constraints enforced.
Build the model generator function

Each table in your MES schema needs its own Pydantic model class. The generate_model_code() function reads the column metadata you pulled from information_schema and emits a complete class definition as a string.

  • In schemas_generator.py, first add a module docstring at the very top of the file (above the existing imports):
"""\nAuto-generates Pydantic models from Unity Catalog information_schema.\nRun this script to regenerate schemas.py from the live DDL.\n"""
  • Below your existing table_name_to_class_name() function, add the model generator function:
def generate_model_code(table_name: str, columns: list) -> str:
    """Generate a Pydantic model class for a single table."""
    class_name = table_name_to_class_name(table_name)
    lines = [f"class {class_name}(BaseModel):"]
    lines.append(f'    \"\"\"Model for {CATALOG}.{SCHEMA}.{table_name}\"\"\"')
    lines.append("")
    for col in columns:
        python_type = map_sql_type(col["data_type"])
        field_name = col["column_name"]
        if col["is_nullable"]:
            type_annotation = f"Optional[{python_type}]"
            default = " = None"
        else:
            type_annotation = python_type
            default = ""
        if field_name in FIELD_VALIDATORS:
            constraints = FIELD_VALIDATORS[field_name]
            constraint_parts = [f"{k}={v}" for k, v in constraints.items()]
            lines.append(
                f"    {field_name}: {type_annotation} = "
                f"Field({', '.join(constraint_parts)})"
            )
        else:
            lines.append(f"    {field_name}: {type_annotation}{default}")
    lines.append("")
    lines.append(f'    class Config:')
    lines.append(f'        extra = \"forbid\"')
    lines.append("")
    return "\n".join(lines)

What does this code do?

  • For each column, map_sql_type() converts the Databricks SQL type (STRING, BIGINT, TIMESTAMP, etc.) to the corresponding Python type using the TYPE_MAP you defined in Step 1.
  • Nullable columns get wrapped in Optional[] with a default of None. Non-nullable columns have no default, so Pydantic requires a value at instantiation.
  • Fields matching FIELD_VALIDATORS (like mde_pct, alpha, power, split_ratio) get B2B-specific constraints via Field(ge=..., le=...). For example, mde_pct is constrained to 0.05-0.50, preventing underpowered or unrealistic effect sizes.
  • The Config class with extra = "forbid" means any field not in the schema raises a validation error. This catches typos and unexpected columns immediately.
  • Save schemas_generator.py.
  • Verify the function is importable by running this quick test:
python3 -c "from schemas_generator import generate_model_code; print(generate_model_code('experiments', [{'column_name': 'experiment_id', 'data_type': 'STRING', 'is_nullable': False}, {'column_name': 'mde_pct', 'data_type': 'DOUBLE', 'is_nullable': False}]))"

You should see a class definition printed with Experiments(BaseModel) containing an experiment_id: str field and an mde_pct field using Field(ge=0.05, le=0.5).

Seeing an ImportError?

  • Make sure you are running the command from the same directory where schemas_generator.py lives.
  • Check that the function is at the module level (not accidentally indented inside another function).
  • Confirm you saved the file before running the test command.

Help me debug an import error when testing generate_model_code in schemas_generator.py.

Assemble the full file generator

The generate_schemas_file() function orchestrates everything. It assembles the file header (imports), generates all 8 enum classes from ENUM_DEFINITIONS, then generates one model class per table.

  • Below generate_model_code(), add the schema file generator function:
def generate_schemas_file(tables: dict) -> str:
    """Generate the complete schemas.py content."""
    header = '''\"\"\"\nAuto-generated Pydantic models for MES Delta tables.\nGenerated from: {catalog}.information_schema.columns\nDO NOT EDIT MANUALLY. Re-run schemas_generator.py to update.\n\"\"\"\nfrom datetime import date, datetime\nfrom decimal import Decimal\nfrom enum import Enum\nfrom typing import Optional\n\nfrom pydantic import BaseModel, Field\n\n\n'''.format(catalog=CATALOG)
    # Generate enum classes
    enum_lines = []
    for enum_name, values in ENUM_DEFINITIONS.items():
        enum_lines.append(f"class {enum_name}(str, Enum):")
        for val in values:
            enum_lines.append(f'    {val.upper()} = \"{val}\"')
        enum_lines.append("")
        enum_lines.append("")
    # Generate model classes
    model_lines = []
    for table_name, columns in sorted(tables.items()):
        model_lines.append(generate_model_code(table_name, columns))
    return header + "\n".join(enum_lines) + "\n" + "\n".join(model_lines)

What does this function produce?

  • The header string contains the docstring plus all necessary Python imports (datetime, Decimal, Enum, Optional, BaseModel, Field) that the generated models need.
  • Enum classes are generated first so the models can reference them. Each enum value gets an uppercase constant name (e.g., PROPOSED = "proposed").
  • Model classes are sorted alphabetically by table name, giving consistent file output regardless of the order information_schema returns results.
  • Save schemas_generator.py.
  • Confirm the function loads without syntax errors by running:
python3 -c "from schemas_generator import generate_schemas_file; print('Schema generator ready')"

You should see Schema generator ready printed with no errors.

Seeing a SyntaxError in the header string?

  • The header uses triple single-quotes to wrap the multi-line string. Make sure you have exactly three single quotes at the start and three at the end before .format(catalog=CATALOG).
  • Check that the .format() call uses catalog (lowercase) inside the string placeholder and CATALOG (uppercase) as the argument.

Help me fix a SyntaxError in my generate_schemas_file header string.

Run the generator and verify your models

The last piece is the entry point that ties introspection to file generation. When you run schemas_generator.py directly, it queries your live schema and writes schemas.py to disk.

  • At the bottom of schemas_generator.py, add the main execution block:
if __name__ == "__main__":
    warehouse_id = os.environ.get("DATABRICKS_WAREHOUSE_ID")
    if not warehouse_id:
        raise ValueError(
            "Set DATABRICKS_WAREHOUSE_ID environment variable"
        )
    print(f"Introspecting {CATALOG}.{SCHEMA}...")
    tables = get_table_schemas(warehouse_id)
    print(f"Found {len(tables)} tables")
    schemas_code = generate_schemas_file(tables)
    output_path = os.path.join(
        os.path.dirname(__file__), "schemas.py"
    )
    with open(output_path, "w") as f:
        f.write(schemas_code)
    print(f"Generated schemas.py with {len(tables)} models")

What does this block do?

  • It reads DATABRICKS_WAREHOUSE_ID from your environment (the same variable you set in Step 1).
  • It calls get_table_schemas() to pull live column metadata from information_schema.
  • It passes the table dict through generate_schemas_file() to produce the full Python source code, then writes it to schemas.py in the same directory.
  • Save schemas_generator.py.
  • Generate your models by running the script:
python3 schemas_generator.py

You should see output like:

Introspecting marketing_experimentation.mes_dev...
Found 17 tables
Generated schemas.py with 17 models
  • Confirm schemas.py was created by running:
ls schemas.py

You should see schemas.py listed in the output.

Now for the real test. Before you run the next command, predict: what do you think happens when you try to create an Experiment with mde_pct=0.03?

  • Test that your generated models enforce B2B constraints by running:
python3 -c "from schemas import Experiment; Experiment(mde_pct=0.03)"

You see a ValidationError because 0.03 is below the 0.05 minimum you defined in FIELD_VALIDATORS. That constraint is now enforced automatically in every piece of code that touches experiment data.

  • Verify the key model imports all work cleanly:
python3 -c "from schemas import Experiment, Metric, DesignDecision, Readout; print('All core models imported successfully')"

You should see All core models imported successfully with no errors.

Seeing a ModuleNotFoundError for pydantic?

  • Confirm your virtual environment is activated. The Pydantic install from Step 1 is only available inside the environment where you ran pip install.
  • If the import fails on a specific model name (e.g., DesignDecision), open schemas.py and check whether the class exists. The class name is PascalCase derived from the table name (design_decisions becomes DesignDecisions). Adjust your import to match the generated name.
  • If you see a ValidationError on import (not on instantiation), a default value in the generated model may violate a constraint. Re-run the generator and check FIELD_VALIDATORS boundaries.

Help me debug import issues with my auto-generated schemas.py Pydantic models.

✔️ Awesome, I've got everything!

Great. Confirm you have saved schemas_generator.py with all three new functions and the __main__ block, and that schemas.py exists in your project directory.

ⓧ I'd like to double check the full code

Scroll up and confirm your schemas_generator.py contains these sections in order:

  • Module docstring at the top.
  • Imports (WorkspaceClient, os).
  • Constants: CATALOG, SCHEMA, TYPE_MAP, FIELD_VALIDATORS, ENUM_DEFINITIONS.
  • Function: get_table_schemas() (from Step 1).
  • Function: map_sql_type() (from Step 1).
  • Function: table_name_to_class_name() (from Step 1).
  • Function: generate_model_code() (added this step).
  • Function: generate_schemas_file() (added this step).
  • The if __name__ == "__main__": block at the bottom (added this step).

Also confirm that schemas.py exists and opens without errors. It should contain 8 Enum classes followed by 17 BaseModel classes.

Your 17-table schema is now captured as type-safe Python models with B2B guardrails baked in. Next up, you will build a CI gate that compares these models against the live schema on every push, so drift never slips through unnoticed.

Build the Schema Validation CI Gate

Your schemas.py gives you 17 type-safe models that mirror your live tables right now. But schemas drift. A column gets renamed, a table picks up a new field, or someone adds a nullable column that your app assumes is required.

Without an automated check, those mismatches only surface when downstream code breaks in production. In this step, you will build a validation script that compares every Pydantic model against the live Unity Catalog schema on every push, catching drift before it reaches your experiments.

In this step, get ready to:
  • Write a validation function that queries information_schema for each table and compares it against your Pydantic models.
  • Build assertion logic that checks field existence, type compatibility, and non-nullable coverage.
  • Integrate the validator into your CI pipeline and confirm it catches schema drift.
Write the information_schema query function

The validator needs to pull the live column set for any given table. You will build this as a standalone script that your CI pipeline can call independently.

  • Create a new file called schema_validator.py in the same directory as schemas.py.
  • Add the imports, constants, and reverse type map by pasting this code:
"""
CI gate: validates Pydantic models against live Unity Catalog schemas.
Run as part of CI pipeline on every push.
"""
from databricks.sdk import WorkspaceClient
import importlib
import inspect
import os
import sys
from datetime import date, datetime
from decimal import Decimal
from typing import get_type_hints, get_origin, get_args, Optional, Union

CATALOG = "marketing_experimentation"
SCHEMA = "mes_dev"

PYTHON_TO_SQL = {
    str: {"STRING", "VARCHAR", "CHAR"},
    int: {"INT", "BIGINT", "SMALLINT", "TINYINT", "INTEGER"},
    float: {"DOUBLE", "FLOAT"},
    Decimal: {"DECIMAL"},
    bool: {"BOOLEAN"},
    date: {"DATE"},
    datetime: {"TIMESTAMP"},
    list: {"ARRAY"},
    dict: {"MAP", "STRUCT"},
}

What does this code do?

PYTHON_TO_SQL is the reverse of the TYPE_MAP you used in the generator. It maps each Python type to the set of SQL types that are acceptable matches. For example, a Python int field is compatible with INT, BIGINT, SMALLINT, TINYINT, or INTEGER.

  • Next, add the get_live_schema() function below the PYTHON_TO_SQL dictionary:


def get_live_schema(warehouse_id: str, table_name: str) -> dict:
    w = WorkspaceClient()
    query = f"""
    SELECT column_name, data_type, is_nullable
    FROM {CATALOG}.information_schema.columns
    WHERE table_catalog = '{CATALOG}'
      AND table_schema = '{SCHEMA}'
      AND table_name = '{table_name}'
    ORDER BY ordinal_position
    """
    result = w.statement_execution.execute_statement(
        warehouse_id=warehouse_id,
        statement=query
    )
    columns = {}
    for row in result.result.data_array:
        columns[row[0]] = {
            "data_type": row[1],
            "is_nullable": row[2] == "YES",
        }
    return columns

What does this function do?

This is the per-table version of the bulk query you wrote in schemas_generator.py. It queries information_schema for a single table's columns and returns a dictionary keyed by column name, with each value containing the SQL data type and nullability.

The WHERE table_name = '{table_name}' filter makes this query fast. Unity Catalog auto-filters results to objects you have privileges on, so no extra grants are needed.

  • Save schema_validator.py.

Seeing a syntax error on the f-string?

Make sure the triple-quoted f-string uses consistent indentation. The closing """ must be on its own line, aligned with the opening lines of the SQL query.

Also confirm you have the two blank lines between PYTHON_TO_SQL and the function definition.

Build the type-checking and assertion logic

The validator needs three helper functions: one to extract fields from a Pydantic model, one to unwrap Optional[X] annotations down to their base type, and one to check whether a Python type is compatible with a given SQL type.

  • Add these three helper functions below get_live_schema():


def get_model_fields(model_class) -> dict:
    fields = {}
    for name, field_info in model_class.model_fields.items():
        fields[name] = {
            "annotation": field_info.annotation,
            "required": field_info.is_required(),
        }
    return fields


def unwrap_optional(annotation):
    origin = get_origin(annotation)
    if origin is Union:
        args = get_args(annotation)
        non_none = [a for a in args if a is not type(None)]
        if len(non_none) == 1:
            return non_none[0]
    return annotation


def type_compatible(python_type, sql_type: str) -> bool:
    base_sql = sql_type.split("(")[0].upper().strip()
    if base_sql.startswith("ARRAY"):
        base_sql = "ARRAY"
    if base_sql.startswith("MAP"):
        base_sql = "MAP"
    if base_sql.startswith("STRUCT"):
        base_sql = "STRUCT"
    if base_sql.startswith("DECIMAL"):
        base_sql = "DECIMAL"
    acceptable = PYTHON_TO_SQL.get(python_type, set())
    return base_sql in acceptable

What do these helpers do?

  • get_model_fields() uses Pydantic v2's model_fields API to extract every field's annotation and whether it is required.
  • unwrap_optional() strips the Optional wrapper so you can compare the inner type (e.g. Optional[str] becomes str).
  • type_compatible() normalises the SQL type (strips precision like DECIMAL(10,2) to DECIMAL) and checks it against the acceptable set in PYTHON_TO_SQL.
  • Now add the core assertion function, validate_table(), which performs both the forward check (model fields exist in the live table) and the reverse check (non-nullable live columns exist in the model):


def validate_table(model_class, table_name: str, warehouse_id: str) -> list:
    errors = []
    live_columns = get_live_schema(warehouse_id, table_name)
    if not live_columns:
        errors.append(f"Table '{table_name}' not found in {CATALOG}.{SCHEMA}")
        return errors
    model_fields = get_model_fields(model_class)
    for field_name, field_info in model_fields.items():
        if field_name not in live_columns:
            errors.append(
                f"Field '{field_name}' in model {model_class.__name__} does not exist in table {CATALOG}.{SCHEMA}.{table_name}"
            )
            continue
        python_type = unwrap_optional(field_info["annotation"])
        sql_type = live_columns[field_name]["data_type"]
        if not type_compatible(python_type, sql_type):
            errors.append(
                f"Type mismatch for '{field_name}' in {model_class.__name__}: model has {python_type.__name__}, table has {sql_type}"
            )
    for col_name, col_info in live_columns.items():
        if not col_info["is_nullable"] and col_name not in model_fields:
            errors.append(
                f"Required column '{col_name}' in table {table_name} is missing from model {model_class.__name__}"
            )
    return errors

What does validate_table check?

This function performs two directional checks that together guarantee full schema coverage.

  • Forward check: every field in the Pydantic model must exist in the live table with a compatible type. This catches stale model fields that reference dropped or renamed columns.
  • Reverse check: every non-nullable column in the live table must have a corresponding model field. This catches new required columns that the model does not account for.
  • Save schema_validator.py.

Unsure about the f-string error messages?

The long f-strings in errors.append() must each be a single string on one line (or wrapped in parentheses across lines). Make sure the opening f" and closing " are on the same logical line.

Add the orchestrator and integrate with CI

The final piece is a run_validation() function that discovers all Pydantic models in schemas.py, converts each class name to a table name, and validates them all. A __main__ block makes the script callable from your CI pipeline.

  • Add the orchestrator functions below validate_table():


def class_name_to_table_name(class_name: str) -> str:
    result = []
    for i, char in enumerate(class_name):
        if char.isupper() and i > 0:
            result.append("_")
        result.append(char.lower())
    return "".join(result)


def run_validation(warehouse_id: str) -> bool:
    sys.path.insert(0, os.path.dirname(__file__))
    import schemas
    from pydantic import BaseModel
    models = []
    for name, obj in inspect.getmembers(schemas):
        if (inspect.isclass(obj)
                and issubclass(obj, BaseModel)
                and obj is not BaseModel):
            models.append((name, obj))
    all_errors = []
    for class_name, model_class in models:
        table_name = class_name_to_table_name(class_name)
        errors = validate_table(model_class, table_name, warehouse_id)
        if errors:
            all_errors.extend(errors)
            print(f"  FAIL: {class_name} ({len(errors)} errors)")
            for err in errors:
                print(f"    - {err}")
        else:
            print(f"  PASS: {class_name}")
    print(f"\n{'='*60}")
    print(f"Results: {len(models) - len(all_errors)} passed, {len(all_errors)} errors")
    return len(all_errors) == 0

What does this code do?

  • class_name_to_table_name() reverses the PascalCase conversion. DesignDecision becomes design_decision, matching your Delta table names.
  • run_validation() uses inspect.getmembers() to discover every BaseModel subclass in schemas.py automatically. No hardcoded model list to maintain.
  • The function returns True only if zero errors are found across all models. Your CI pipeline uses this exit code to gate merges.
  • Finally, add the __main__ entry point at the bottom of the file:


if __name__ == "__main__":
    warehouse_id = os.environ.get("DATABRICKS_WAREHOUSE_ID")
    if not warehouse_id:
        raise ValueError("Set DATABRICKS_WAREHOUSE_ID environment variable")
    print(f"Validating schemas against {CATALOG}.{SCHEMA}...")
    print("=" * 60)
    success = run_validation(warehouse_id)
    sys.exit(0 if success else 1)

How does this fit into CI?

The script exits with code 0 on success and 1 on failure. Add it as a step in your existing CI pipeline (the same one that runs your statistical engine's 14-test suite). Any push that introduces schema drift will fail the pipeline before it reaches production.

  • Save schema_validator.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved schema_validator.py and that it contains all six chunks in order.

ⓧ I'd like to double check the full code

Scroll up and confirm your file contains these sections in order:

  • Module docstring, imports, CATALOG/SCHEMA constants, and PYTHON_TO_SQL dictionary.
  • get_live_schema() function.
  • get_model_fields(), unwrap_optional(), and type_compatible() helpers.
  • validate_table() function with forward and reverse checks.
  • class_name_to_table_name() and run_validation() functions.
  • if __name__ == "__main__": block at the bottom.

Now run the validator against your live schema to confirm all 17 models pass.

Before you run this, what do you expect the output to look like? How many PASS lines should you see?

  • Run the schema validator by executing:
python schema_validator.py

You should see 17 PASS lines (one per model) and zero errors in the results summary.

Seeing import errors or FAIL lines?

  • If you get ModuleNotFoundError: No module named 'schemas', confirm schemas.py is in the same directory as schema_validator.py.
  • If a model shows FAIL with a type mismatch, your schemas.py may be stale. Re-run python schemas_generator.py to regenerate it from the live schema, then try the validator again.
  • If you see DATABRICKS_WAREHOUSE_ID errors, confirm the environment variable is still set in your current terminal session.

Why is my schema validator failing on models that should match?

Now test the CI gate's ability to catch drift. This is the scenario you are protecting against: someone adds a field to a model that does not exist in the live table.

  • Open schemas.py and find the Experiment model class (one of the 17 auto-generated classes near the top of the models section).
  • Add a fake field anywhere inside the class body:
    fake_field: str = "should_not_exist"

Before you re-run the validator, what do you think will happen? Will it still pass, or will it catch the fake field?

  • Run the validator again:
python schema_validator.py

The validator catches the drift. You see a FAIL line for the Experiment model with the message:

Field 'fake_field' in model Experiment does not exist in table marketing_experimentation.mes_dev.experiments

The script also exits with code 1, which would fail your CI pipeline and block the merge.

  • Remove the fake_field line from schemas.py and save the file.
  • Run the validator one final time to confirm you are back to a clean state with all 17 models passing.

Your schema contract is now enforced on every push. Any drift between your Pydantic models and the live Delta tables gets caught before it can corrupt downstream experiments. Next up, you will populate those models with real baseline data from your source tables.

Populate Real Metric Baselines

Your CI gate now catches any schema drift between your Pydantic models and the live tables. The code-layer contract is solid. But open your metrics_dictionary table and you'll see the problem: every baseline_value and baseline_variance is still NULL or demo data.

Every downstream calculation depends on accurate baselines. Power calculators use them for sample size estimates. The wizard uses them for sensitivity tables. Opportunity cost formulas need real conversion rates and ACV. Placeholder values produce misleading power estimates that either underpower your tests or waste resources.

In this step, get ready to:
  • Write baseline extraction queries that pull real values from source tables, segmented by account segment.
  • Execute the extractions and populate the metrics_dictionary with real baseline_value, baseline_variance, and baseline_updated_at.
  • Pull real accrual rates and ACV per segment from CRM data into lookup tables.
Write baseline extraction queries

You need a script that reads each metric's definition from metrics_dictionary, executes the appropriate statistical computation against the live source tables, and writes the result back. The computation differs by metric type: proportions use p*(1-p) for variance, while continuous metrics use sample variance directly.

  • Create a new file called baseline_extractor.py in your project directory.
  • Add the file setup with imports, catalog constants, and the segment list by pasting this code:
"""
Extracts real metric baselines from source tables and populates
the metrics_dictionary Delta table.
"""
from databricks.sdk import WorkspaceClient
from datetime import date
import os

CATALOG = "marketing_experimentation"
SCHEMA = "mes_dev"
SEGMENTS = ["Enterprise", "Mid-Market", "SMB"]

What does this set up?

The SEGMENTS list defines the three B2B account tiers your metrics are segmented by. Every baseline extraction runs once per segment, so you get separate power calculations for Enterprise, Mid-Market, and SMB cohorts.

  • Add the get_metrics() function below the constants. This loads every active metric's definition from your metrics_dictionary:


def get_metrics(warehouse_id: str) -> list:
    """Load all active metrics from metrics_dictionary."""
    w = WorkspaceClient()
    query = f"""
    SELECT metric_id, metric_name, metric_type, sql_definition,
           numerator_table, numerator_column,
           denominator_table, denominator_column,
           minimum_sample_warning
    FROM {CATALOG}.{SCHEMA}.metrics_dictionary
    WHERE deprecated_at IS NULL
    """
    result = w.statement_execution.execute_statement(
        warehouse_id=warehouse_id,
        statement=query
    )
    metrics = []
    for row in result.result.data_array:
        metrics.append({
            "metric_id": row[0],
            "metric_name": row[1],
            "metric_type": row[2],
            "sql_definition": row[3],
            "numerator_table": row[4],
            "numerator_column": row[5],
            "denominator_table": row[6],
            "denominator_column": row[7],
            "minimum_sample_warning": int(row[8]) if row[8] else 100,
        })
    return metrics

What does this function do?

This queries metrics_dictionary for every non-deprecated metric. It pulls the numerator_table and numerator_column so the extraction functions know where to find the raw data. The minimum_sample_warning threshold prevents writing baselines from insufficient data (defaults to 100 if not set).

Now you need the two computation functions. Proportion metrics (like conversion rates) compute variance as p*(1-p). Continuous metrics (like ACV or deal cycle time) use sample variance directly.

  • Add the compute_proportion_baseline() function. This builds a query that computes the conversion rate, standard error, and sample size from the source table, filtered to a 90-day lookback window:


def compute_proportion_baseline(
    warehouse_id: str, metric: dict, segment: str
) -> dict:
    """Compute baseline for a proportion metric."""
    w = WorkspaceClient()
    query = f"""
    SELECT
        SUM(CAST({metric['numerator_column']} AS DOUBLE))
            / COUNT(*) AS baseline_value,
        SQRT(
            (SUM(CAST({metric['numerator_column']} AS DOUBLE))
            / COUNT(*))
            * (1 - SUM(CAST({metric['numerator_column']} AS DOUBLE))
            / COUNT(*))
            / COUNT(*)
        ) AS baseline_se,
        COUNT(*) AS sample_size
    FROM {metric['numerator_table']}
    WHERE account_segment = '{segment}'
      AND created_at >= DATEADD(DAY, -90, CURRENT_DATE())
    """
  • Directly below the query string (still inside the same function), add the result processing that executes the query, extracts the values, and returns the baseline dictionary:
    result = w.statement_execution.execute_statement(
        warehouse_id=warehouse_id,
        statement=query
    )
    row = result.result.data_array[0]
    baseline_value = float(row[0]) if row[0] else None
    baseline_se = float(row[1]) if row[1] else None
    sample_size = int(row[2]) if row[2] else 0

    if baseline_value is None or sample_size == 0:
        return None

    # Variance for a proportion: p*(1-p)
    baseline_variance = baseline_value * (1 - baseline_value)

    return {
        "baseline_value": baseline_value,
        "baseline_variance": baseline_variance,
        "sample_size": sample_size,
        "segment": segment,
    }

Why p*(1-p) for variance?

For a binary outcome (converted or not), the variance of a proportion is p*(1-p). This is what your power calculator needs to determine how many accounts you need in each arm. The 90-day lookback window ensures your baseline reflects recent behaviour, not stale historical patterns.

  • Add the compute_continuous_baseline() function below. This handles metrics like ACV or deal velocity where you need sample mean and variance:


def compute_continuous_baseline(
    warehouse_id: str, metric: dict, segment: str
) -> dict:
    """Compute baseline for a continuous metric."""
    w = WorkspaceClient()
    query = f"""
    SELECT
        AVG(CAST({metric['numerator_column']} AS DOUBLE))
            AS baseline_value,
        VARIANCE(CAST({metric['numerator_column']} AS DOUBLE))
            AS baseline_variance,
        COUNT(*) AS sample_size
    FROM {metric['numerator_table']}
    WHERE account_segment = '{segment}'
      AND created_at >= DATEADD(DAY, -90, CURRENT_DATE())
    """
  • Add the result processing for the continuous function directly below:
    result = w.statement_execution.execute_statement(
        warehouse_id=warehouse_id,
        statement=query
    )
    row = result.result.data_array[0]
    baseline_value = float(row[0]) if row[0] else None
    baseline_variance = float(row[1]) if row[1] else None
    sample_size = int(row[2]) if row[2] else 0

    if baseline_value is None or sample_size == 0:
        return None

    return {
        "baseline_value": baseline_value,
        "baseline_variance": baseline_variance,
        "sample_size": sample_size,
        "segment": segment,
    }

How does this differ from the proportion function?

Instead of computing p*(1-p), continuous metrics use the SQL VARIANCE() aggregate directly. This gives you the sample variance needed for power calculations on metrics like average deal size or time-to-close.

  • Save baseline_extractor.py.
  • Confirm the file imports cleanly by running:
python -c "from baseline_extractor import get_metrics, compute_proportion_baseline, compute_continuous_baseline; print('Imports OK')"

You should see Imports OK printed with no errors.

Seeing an ImportError?

Check that your function definitions are at the module level (not indented under another block). Verify that the from databricks.sdk import WorkspaceClient line is at the top of the file.

Help me debug my baseline_extractor.py import error

Execute the extractions and populate metrics_dictionary

The query functions are ready. Now you need the orchestration layer: a function to write baselines back to the Delta table, and a main loop that iterates every metric across every segment.

  • Add the update_baseline() function below your compute functions. This writes a computed baseline back to metrics_dictionary:


def update_baseline(
    warehouse_id: str, metric_id: str, baseline: dict
) -> None:
    """Write computed baseline back to metrics_dictionary."""
    w = WorkspaceClient()
    today = date.today().isoformat()
    query = f"""
    UPDATE {CATALOG}.{SCHEMA}.metrics_dictionary
    SET baseline_value = {baseline['baseline_value']},
        baseline_variance = {baseline['baseline_variance']},
        baseline_updated_at = '{today}'
    WHERE metric_id = '{metric_id}'
    """
    w.statement_execution.execute_statement(
        warehouse_id=warehouse_id,
        statement=query
    )

What does this function do?

This issues an UPDATE against your Delta table, setting the three baseline columns and stamping today's date as baseline_updated_at. Delta's time travel means any erroneous write can be rolled back within the 30-day retention window.

  • Add the first half of the extract_all_baselines() orchestration function. This loops every metric across all three segments and dispatches to the correct compute function:


def extract_all_baselines(warehouse_id: str) -> dict:
    """Extract baselines for all metrics across all segments."""
    metrics = get_metrics(warehouse_id)
    results = {"success": [], "failed": [], "skipped": []}

    for metric in metrics:
        print(f"Processing: {metric['metric_name']}...")

        for segment in SEGMENTS:
            try:
                if metric["metric_type"] == "proportion":
                    baseline = compute_proportion_baseline(
                        warehouse_id, metric, segment
                    )
                elif metric["metric_type"] in ("continuous", "ratio"):
                    baseline = compute_continuous_baseline(
                        warehouse_id, metric, segment
                    )
                else:
                    results["skipped"].append(
                        f"{metric['metric_name']} ({segment}): "
                        f"unsupported type {metric['metric_type']}"
                    )
                    continue
  • Continue the function by adding the validation checks and the update call. This goes directly below the continue statement, still inside the try block:

                if baseline is None:
                    results["skipped"].append(
                        f"{metric['metric_name']} ({segment}): "
                        f"no data returned"
                    )
                    continue

                if baseline["sample_size"] < metric[
                    "minimum_sample_warning"
                ]:
                    results["skipped"].append(
                        f"{metric['metric_name']} ({segment}): "
                        f"sample size {baseline['sample_size']} below "
                        f"minimum {metric['minimum_sample_warning']}"
                    )
                    continue

                update_baseline(
                    warehouse_id, metric["metric_id"], baseline
                )
                results["success"].append(
                    f"{metric['metric_name']} ({segment}): "
                    f"baseline={baseline['baseline_value']:.4f}, "
                    f"var={baseline['baseline_variance']:.6f}, "
                    f"n={baseline['sample_size']}"
                )
  • Close the function with the exception handler and return statement. Add this directly below the success append:

            except Exception as e:
                results["failed"].append(
                    f"{metric['metric_name']} ({segment}): {str(e)}"
                )

    return results

Why three result buckets?

The function categorises each metric-segment pair into success, skipped, or failed. Skipped means the data was insufficient (too few rows or NULL results). Failed means an actual error occurred. This lets you triage issues without re-running the entire extraction.

  • Add the __main__ block at the bottom of the file to make the script runnable:


if __name__ == "__main__":
    warehouse_id = os.environ.get("DATABRICKS_WAREHOUSE_ID")
    if not warehouse_id:
        raise ValueError(
            "Set DATABRICKS_WAREHOUSE_ID environment variable"
        )

    print("Extracting real baselines for all metrics...")
    print("=" * 60)

    results = extract_all_baselines(warehouse_id)

    print(f"\nSuccess: {len(results['success'])}")
    for s in results["success"]:
        print(f"  {s}")

    print(f"\nSkipped: {len(results['skipped'])}")
    for s in results["skipped"]:
        print(f"  {s}")

    print(f"\nFailed: {len(results['failed'])}")
    for f in results["failed"]:
        print(f"  {f}")
  • Save baseline_extractor.py.
  • Run the extractor against your live warehouse:
python baseline_extractor.py

Before you check the output, predict: do you expect any metrics to land in the skipped bucket? Think about which segments might have fewer than 100 accounts in the 90-day window.

You should see output listing each metric being processed, followed by a summary showing how many succeeded, were skipped, and failed. Most metrics should show as success with real baseline values and variance estimates.

Seeing all metrics in the failed bucket?

Verify your DATABRICKS_WAREHOUSE_ID environment variable is still set correctly. Check that the source tables referenced in your metrics_dictionary (numerator_table column) exist and contain an account_segment column.

Help me debug why my baseline extraction is failing for all metrics

Pull real accrual rates and ACV from CRM

Your opportunity cost formulas also need real accrual rates (accounts entering each channel per day) and ACV per segment. These live in lookup_channel_method and feed the holdout cost calculations.

  • Run the following query in a Databricks notebook or SQL editor to populate your lookup table with real accrual rates and ACV pulled from CRM data:
-- Update lookup_channel_method with real accrual rates from CRM
UPDATE marketing_experimentation.mes_dev.lookup_channel_method lc
SET
    lc.accounts_per_day = src.daily_account_rate,
    lc.acv_per_segment = src.avg_acv
FROM (
    SELECT
        channel,
        account_segment,
        COUNT(DISTINCT account_id) / 90.0 AS daily_account_rate,
        AVG(annual_contract_value) AS avg_acv
    FROM marketing_experimentation.mes_dev.crm_accounts
    WHERE created_at >= DATEADD(DAY, -90, CURRENT_DATE())
    GROUP BY channel, account_segment
) src
WHERE lc.channel = src.channel
  AND lc.account_segment = src.account_segment

What does this query do?

This computes the average daily account inflow per channel and the average ACV per segment from your CRM data over the last 90 days. The accrual rate tells your opportunity cost calculator how quickly accounts accumulate in a holdout. The ACV tells it how much revenue each held-out account represents.

  • Verify the lookup table was updated by running this query:
SELECT channel, account_segment, accounts_per_day, acv_per_segment
FROM marketing_experimentation.mes_dev.lookup_channel_method
WHERE accounts_per_day IS NOT NULL
ORDER BY channel, account_segment

You should see rows for each channel-segment combination with non-null accrual rates and ACV values.

Seeing NULL values in the lookup table?

Make sure the channel and account_segment values in your CRM source table match the values stored in lookup_channel_method exactly (case-sensitive). If the JOIN finds no matches, no rows are updated.

Help me debug why my lookup_channel_method UPDATE did not populate values

Now confirm the end-to-end result. Before running this verification query, predict: will every metric row have a non-null baseline, or do you expect some gaps?

  • Run the verification query against metrics_dictionary:
SELECT metric_name, baseline_value, baseline_variance, baseline_updated_at
FROM marketing_experimentation.mes_dev.metrics_dictionary
WHERE baseline_updated_at IS NOT NULL

Every metric that had sufficient data should now show a non-null baseline_value with today's date in baseline_updated_at. Spot-check 2-3 metrics manually against their source tables to confirm the values match.

✔️ Awesome, I've got everything!

Confirm you have saved baseline_extractor.py and that your metrics_dictionary shows real baselines.

ⓧ I'd like to double check the full code

Your baseline_extractor.py should contain these sections in order. Scroll up and confirm each is present:

  • Module docstring, imports (WorkspaceClient, date, os), and constants (CATALOG, SCHEMA, SEGMENTS).
  • get_metrics() function querying metrics_dictionary.
  • compute_proportion_baseline() function with SUM/COUNT query and p*(1-p) variance.
  • compute_continuous_baseline() function with AVG/VARIANCE query.
  • update_baseline() function issuing an UPDATE statement.
  • extract_all_baselines() function orchestrating the loop with success/skipped/failed tracking.
  • if __name__ == "__main__": block that runs the extraction and prints results.

Your metrics_dictionary now holds real baselines derived from live source data. Every power calculation, sensitivity table, and opportunity cost estimate downstream will use these values instead of placeholders. Next up, you'll validate that all metric SQL definitions actually execute cleanly against those source tables.

Validate All Metric SQL Definitions

Your metrics_dictionary now has real baselines, and your Pydantic models enforce the schema contract. There is one gap left: you have no proof that the SQL definitions stored in each metric row actually execute against the live source tables.

A metric definition that fails silently (wrong table reference, stale column name, permission issue) means your analysis runbooks will break at readout time rather than at design time. In this step, you will build a validation harness that executes every metric's sql_definition against your live tables and confirms they produce real results.

In this step, get ready to:
  • Build a metric SQL validator that executes each metric definition with a LIMIT 1000 and checks for valid output.
  • Run the validator across all metrics and review the pass/fail/warning results table.
  • Diagnose and fix any failures until every metric shows a passed status.
Create the metric validator file

This script will query every active metric's SQL definition, execute it against the live tables, and report whether each one returns valid data.

  • Create a new file called metric_validator.py in the same directory as your other MES scripts.
  • Add the file header with imports and constants:
"""
Validates all metric SQL definitions in metrics_dictionary
execute cleanly against live source tables.
"""
from databricks.sdk import WorkspaceClient
import os
import time

CATALOG = "marketing_experimentation"
SCHEMA = "mes_dev"

What does this set up?

The time import lets you measure how long each metric query takes to execute. The catalog and schema constants point at the same marketing_experimentation.mes_dev location your other scripts use.

  • Add the get_all_metrics function below the constants:
def get_all_metrics(warehouse_id: str) -> list:
    """Load all active metrics with their SQL definitions."""
    w = WorkspaceClient()
    query = f"""
    SELECT metric_id, metric_name, metric_type, sql_definition,
           baseline_value, baseline_variance
    FROM {CATALOG}.{SCHEMA}.metrics_dictionary
    WHERE deprecated_at IS NULL
      AND sql_definition IS NOT NULL
    """
    result = w.statement_execution.execute_statement(
        warehouse_id=warehouse_id,
        statement=query
    )
    metrics = []
    for row in result.result.data_array:
        metrics.append({
            "metric_id": row[0],
            "metric_name": row[1],
            "metric_type": row[2],
            "sql_definition": row[3],
            "baseline_value": float(row[4]) if row[4] else None,
            "baseline_variance": float(row[5]) if row[5] else None,
        })
    return metrics

What does this function do?

This pulls every active (non-deprecated) metric that has a SQL definition from your metrics_dictionary. It also grabs the stored baseline values so you can compare them against what the SQL actually computes.

The sql_definition IS NOT NULL filter ensures you only try to validate metrics that actually have a query to run.

  • Now add the first half of the validate_metric_sql function. This sets up the result record and constructs the test query:
def validate_metric_sql(warehouse_id: str, metric: dict) -> dict:
    """Execute a metric's SQL definition and validate the output."""
    w = WorkspaceClient()
    result_record = {
        "metric_id": metric["metric_id"],
        "metric_name": metric["metric_name"],
        "status": "unknown",
        "row_count": 0,
        "execution_time_ms": 0,
        "error": None,
        "sanity_check": None,
    }

    # Wrap the SQL with LIMIT to prevent runaway queries
    test_sql = f"""
    SELECT * FROM (
        {metric['sql_definition']}
    ) AS metric_output
    LIMIT 1000
    """

Why wrap with LIMIT 1000?

Some metric definitions scan large fact tables. Wrapping in a subquery with LIMIT 1000 proves the SQL executes and returns data without burning warehouse compute on a full table scan.

The result_record dictionary tracks everything about this validation attempt: status, row count, execution time, and any error message.

  • Add the execution logic directly below the test_sql definition, still inside validate_metric_sql:
    start_time = time.time()
    try:
        result = w.statement_execution.execute_statement(
            warehouse_id=warehouse_id,
            statement=test_sql
        )
        elapsed_ms = int((time.time() - start_time) * 1000)
        result_record["execution_time_ms"] = elapsed_ms

        if result.result and result.result.data_array:
            row_count = len(result.result.data_array)
            result_record["row_count"] = row_count

            if row_count == 0:
                result_record["status"] = "warning"
                result_record["error"] = "Query returned 0 rows"
            else:
                result_record["status"] = "passed"
                result_record["sanity_check"] = "rows_returned"
        else:
            result_record["status"] = "warning"
            result_record["error"] = "No result data returned"

    except Exception as e:
        elapsed_ms = int((time.time() - start_time) * 1000)
        result_record["execution_time_ms"] = elapsed_ms
        result_record["status"] = "failed"
        result_record["error"] = str(e)

    return result_record

What does the execution logic check?

  • If the query returns rows, the metric gets status passed. The SQL definition is valid against the live tables.
  • If the query runs but returns zero rows or null data, it gets status warning. The SQL is syntactically valid but may reference empty data.
  • If the query throws an exception (wrong table, bad column, permission denied), it gets status failed. This is the category that needs fixing.
  • Save metric_validator.py.

Indentation looks off?

The execution block (starting with start_time = time.time()) must be indented at the same level as the result_record and test_sql assignments above it. All three are inside validate_metric_sql.

Help me fix indentation in my validate_metric_sql function.

Add the orchestrator and run the suite

The validator function handles one metric at a time. You need an orchestrator that loops through every metric and prints a formatted results table.

  • Add the run_full_validation function below validate_metric_sql:
def run_full_validation(warehouse_id: str) -> list:
    """Validate all metric SQL definitions."""
    metrics = get_all_metrics(warehouse_id)
    results = []

    print(f"Validating {len(metrics)} metric SQL definitions...")
    print("-" * 70)
    print(f"{'Metric':<35} {'Status':<10} {'Rows':<8} {'Time(ms)':<10}")
    print("-" * 70)

    for metric in metrics:
        result = validate_metric_sql(warehouse_id, metric)
        results.append(result)

        status_symbol = {
            "passed": "PASS",
            "warning": "WARN",
            "failed": "FAIL",
        }.get(result["status"], "????")

        print(
            f"{result['metric_name']:<35} "
            f"{status_symbol:<10} "
            f"{result['row_count']:<8} "
            f"{result['execution_time_ms']:<10}"
        )

        if result["error"]:
            print(f"  Error: {result['error']}")

What does this function do?

This loops through every active metric from your dictionary, runs the SQL validator on each one, and prints a formatted table row showing the metric name, pass/fail status, how many rows came back, and how long the query took.

  • Add the summary section and main block directly below the for loop. This completes the file:
    # Summary
    passed = sum(1 for r in results if r["status"] == "passed")
    warned = sum(1 for r in results if r["status"] == "warning")
    failed = sum(1 for r in results if r["status"] == "failed")

    print("-" * 70)
    print(f"Total: {len(results)} | Passed: {passed} | "
          f"Warnings: {warned} | Failed: {failed}")

    return results


if __name__ == "__main__":
    warehouse_id = os.environ.get("DATABRICKS_WAREHOUSE_ID")
    if not warehouse_id:
        raise ValueError(
            "Set DATABRICKS_WAREHOUSE_ID environment variable"
        )

    results = run_full_validation(warehouse_id)

    # Exit with error code if any failures
    failures = [r for r in results if r["status"] == "failed"]
    if failures:
        print(f"\n{len(failures)} metric(s) FAILED validation.")
        exit(1)
    else:
        print("\nAll metrics validated successfully.")
        exit(0)

What does the main block do?

The summary counts how many metrics passed, warned, or failed. The main block exits with code 1 if any metric failed, making it suitable as a CI gate. A non-zero exit code fails the pipeline.

  • Save metric_validator.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved the file before running it.

ⓧ I'd like to double check the full code

Your metric_validator.py should contain these sections in order. Scroll up and confirm each is present:

  • File header: docstring, three imports, CATALOG and SCHEMA constants.
  • get_all_metrics function: queries metrics_dictionary for active metrics with SQL definitions.
  • validate_metric_sql function: builds a LIMIT 1000 wrapper, executes, and returns a result_record dict with status/row_count/execution_time_ms/error.
  • run_full_validation function: loops all metrics, prints formatted table, prints summary counts.
  • Main block: calls run_full_validation and exits with code 1 on failures.

Before you run the validator, what do you expect to see? Will every metric pass cleanly on the first run, or do you think some SQL definitions might reference stale tables or columns?

  • Run the full metric validation suite:
python metric_validator.py

You will see a formatted table listing every metric with its status, row count, and execution time. The summary line at the bottom shows total passed, warnings, and failed counts.

Seeing connection or warehouse errors?

Confirm your DATABRICKS_WAREHOUSE_ID environment variable is set correctly. This is the same warehouse ID you used for baseline_extractor.py in the previous step.

Help me debug a connection error when running metric_validator.py against my Databricks SQL warehouse.

Diagnose and fix any failed metrics

If any metrics show FAIL in the output, the error column tells you exactly what broke. Each failure falls into one of three categories.

  • Stale table reference: the sql_definition points at a table that was renamed or moved. Update the table name in the metric's sql_definition column to match the current table.
  • Stale column name: a column referenced in the SQL no longer exists in the source table. Check your schemas.py for the correct column name and update the sql_definition.
  • Permission issue: the service principal or your user lacks SELECT access on a referenced table. Grant the appropriate Unity Catalog privilege on the source table.
  • For each failed metric, fix the sql_definition in your metrics_dictionary table using an UPDATE statement in your Databricks notebook.
  • After applying fixes, re-run the validator:
python metric_validator.py
  • Repeat until the summary line shows zero in the Failed column and the script exits with code 0.

What should I see?

Your final output shows a table of all metrics with PASS status, row counts above zero, and execution times in milliseconds. The last line reads: All metrics validated successfully. This confirms every metric SQL definition in your dictionary runs cleanly against your live source tables.

🙋‍♀️ Metrics stuck in warning state?

A WARN status means the SQL executed successfully but returned zero rows. This usually means the WHERE clause is too restrictive for the current data (e.g., a date filter that excludes all rows). Check whether the source table has data for the time window the SQL references.

Help me debug a metric SQL definition that returns zero rows even though I know the source table has data.

Your entire MES data layer is now production-ready. The Pydantic models enforce the schema contract, the CI gate catches drift, every metric has real baselines, and all SQL definitions run cleanly against live data. This is the gate that unlocks running real experiments through the system.

Secret mission

Build Baseline Drift Detector

Baselines go stale when the market shifts, product changes, or seasonality hits. A 20%+ drift means your power calculations are wrong and experiments designed against the old baseline may be underpowered. Build an automated detector that catches this before it breaks your experiments.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely within your existing Databricks workspace, so there are no new ongoing costs.

Resources you used:

  • schemas_generator.py and schemas.py (auto-generated Pydantic models for all 17 MES tables).
  • schema_validator.py (CI gate that catches schema drift).
  • baseline_extractor.py (populates real baselines into metrics_dictionary).
  • metric_validator.py (validates all metric SQL definitions against live tables).
  • baseline_drift_detector.py and drift_detection_results.json (Secret Mission drift detection script and output).
  • Updated rows in metrics_dictionary and lookup_channel_method Delta tables.

Keep everything running

No action needed. This is production hardening. All artifacts are permanent and required for your MES to function in production.

  • Your Pydantic models and CI gate prevent silent schema drift on every push.
  • Your real baselines power accurate sample size calculations and opportunity cost estimates.
  • Your metric SQL validator confirms the data layer is ready for live experiments.
  • Delta time travel (30-day retention) provides rollback safety for any erroneous baseline write.

Pause - I'll come back to this later

There are no running processes to shut down. Your scripts only execute when you run them manually or via a scheduled Databricks job.

  • If you scheduled the drift detector as a Databricks job, pause it from the Databricks Jobs UI to avoid unnecessary warehouse compute.
  • Your Python scripts and Delta table data remain intact. You can resume at any time.
  • Delta time travel retains 30 days of history, so baseline writes from this session can be rolled back within that window.

Delete - I don't want to use this again

Remove all project scripts and revert Delta table changes. Only do this if you are sure you do not need the production hardening layer.

  • Delete the Python scripts by running:
rm schemas_generator.py schemas.py schema_validator.py baseline_extractor.py metric_validator.py baseline_drift_detector.py drift_detection_results.json
  • Revert the baseline writes in metrics_dictionary using Delta time travel. Run this in a Databricks notebook:
RESTORE TABLE marketing_experimentation.mes_dev.metrics_dictionary
TO VERSION AS OF (SELECT MAX(version) - 1 FROM (DESCRIBE HISTORY marketing_experimentation.mes_dev.metrics_dictionary))
  • Revert the accrual rate and ACV updates in lookup_channel_method with the same approach:
RESTORE TABLE marketing_experimentation.mes_dev.lookup_channel_method
TO VERSION AS OF (SELECT MAX(version) - 1 FROM (DESCRIBE HISTORY marketing_experimentation.mes_dev.lookup_channel_method))
  • If you scheduled the drift detector as a Databricks job, delete the job from the Databricks Jobs UI by selecting the job and clicking Delete.
  • Uninstall Pydantic from your environment by running:
pip uninstall pydantic -y

Nice Work!

Nice Work!

Nice work! You have transformed your Marketing Experimentation System from demo mode to a production-hardened data layer with real baselines, type-safe contracts, and automated validation.

You've learned how to:

  • Auto-generate Pydantic models from live Unity Catalog schemas and enforce B2B-specific validation rules (MDE range, assignment unit defaults, lifecycle enums) at the code level.
  • Build a CI gate that compares Pydantic model definitions against live information_schema on every push, catching schema drift before it reaches production.
  • Populate every metric in metrics_dictionary with real baseline values, variance estimates, and accrual rates segmented by account segment, then validate all metric SQL definitions execute cleanly against live source tables.
  • Secret Mission: Built an automated baseline drift detector ready to run as a scheduled Databricks job, ensuring your baselines never silently go stale.

Ready to quiz yourself?