07-design-validation

name	07-design-validation
description	Cross-validation of Gold layer design artifacts during the design phase. Use when validating that YAML schemas, ERDs, lineage CSVs, and PK/FK references are internally consistent before handing off to implementation. Catches design-time inconsistencies (e.g., column in ERD but not in YAML, FK referencing non-existent table) that would otherwise surface as runtime bugs.
license	Apache-2.0
clients	["ide_cli","genie_code"]
bundle_resource	none
deploy_verb	bundle_deploy
deploy_note	Design-phase pattern; output feeds the Gold design/setup artifacts deployed downstream via `bundle deploy --target dev`. No standalone resource.
coverage	full
metadata	{"author":"prashanth subrahmanyam","version":"1.0.0","domain":"gold","role":"worker","pipeline_stage":1,"pipeline_stage_name":"gold-design","called_by":["gold-layer-design"],"standalone":true,"last_verified":"2026-04-17","volatility":"low","upstream_sources":[]}

Design Consistency Validation

Overview

Gold layer design produces multiple interconnected artifacts — YAML schemas, Mermaid ERDs, column lineage CSVs, and PK/FK constraint definitions. These artifacts are created across different phases and can drift out of sync. This skill provides cross-validation patterns to catch inconsistencies during the design phase, before they propagate into implementation bugs.

Key Principle: Catch design inconsistencies early. A column missing from a YAML schema but present in an ERD is cheap to fix in design; it's a UNRESOLVED_COLUMN runtime error in implementation.

Companion skill: For runtime DataFrame-vs-DDL schema validation during implementation, see pipeline-workers/05-schema-validation/SKILL.md.

When to Use This Skill

• Completing Phase 8 (Design Validation) of the Gold Layer Design workflow
• Cross-checking YAML schemas against ERD diagrams
• Validating that all YAML columns have lineage metadata
• Ensuring PK/FK references point to valid tables and columns
• Running pre-handoff validation before implementation begins

Core Problem: Design Artifact Drift

Gold layer design involves multiple interconnected artifacts:

Artifact	Location	Contains
YAML Schemas	gold_layer_design/yaml/{domain}/	Column names, types, PKs, FKs, descriptions
ERD Diagrams	gold_layer_design/erd_master.md	Entity names, column names, relationships
Lineage CSV	gold_layer_design/COLUMN_LINEAGE.csv	Source → target column mappings
Source Mapping	gold_layer_design/SOURCE_TABLE_MAPPING.csv	Table-level source → Gold mapping

Problem: These are authored at different phases (ERD at Phase 3, YAML at Phase 4, Lineage at Phase 5). Changes in one may not be reflected in others.

Validation 1: YAML ↔ ERD Consistency

What to Check

Every column in the ERD must exist in the corresponding YAML schema, and vice versa.

Validation Pattern

import yaml
import re
from pathlib import Path

def validate_yaml_erd_consistency(yaml_dir: Path, erd_path: Path) -> dict:
    """Cross-check YAML schemas against ERD diagram."""
    
    # Parse YAML schemas
    yaml_tables = {}
    for yaml_file in yaml_dir.rglob("*.yaml"):
        with open(yaml_file) as f:
            spec = yaml.safe_load(f)
        table_name = spec.get("table_name", yaml_file.stem)
        yaml_tables[table_name] = {
            col["name"] for col in spec.get("columns", [])
        }
    
    # Parse ERD (extract entity names and columns from Mermaid syntax)
    erd_tables = {}
    with open(erd_path) as f:
        erd_content = f.read()
    
    # Match Mermaid ERD entity blocks: table_name { ... }
    entity_pattern = re.compile(
        r'(\w+)\s*\{([^}]*)\}', re.MULTILINE | re.DOTALL
    )
    for match in entity_pattern.finditer(erd_content):
        table_name = match.group(1)
        columns_block = match.group(2)
        columns = set()
        for line in columns_block.strip().split('\n'):
            line = line.strip()
            if line and not line.startswith('%%'):
                parts = line.split()
                if len(parts) >= 2:
                    columns.add(parts[1])  # type column_name
        erd_tables[table_name] = columns
    
    # Cross-check
    issues = []
    
    # Tables in YAML but not in ERD
    for table in yaml_tables:
        if table not in erd_tables:
            issues.append(f"YAML table '{table}' missing from ERD")
    
    # Tables in ERD but not in YAML
    for table in erd_tables:
        if table not in yaml_tables:
            issues.append(f"ERD table '{table}' missing from YAML")
    
    # Column-level comparison for common tables
    for table in set(yaml_tables) & set(erd_tables):
        yaml_cols = yaml_tables[table]
        erd_cols = erd_tables[table]
        
        for col in yaml_cols - erd_cols:
            issues.append(f"YAML column '{table}.{col}' missing from ERD")
        for col in erd_cols - yaml_cols:
            issues.append(f"ERD column '{table}.{col}' missing from YAML")
    
    return {
        "valid": len(issues) == 0,
        "issues": issues,
        "yaml_table_count": len(yaml_tables),
        "erd_table_count": len(erd_tables)
    }

Common Mismatches

Mismatch	Cause	Fix
Column in ERD, missing in YAML	ERD updated after YAML was generated	Add column to YAML schema
Column in YAML, missing in ERD	YAML updated without ERD refresh	Add column to ERD or regenerate ERD
Table name differs	Rename in one artifact but not the other	Align names across all artifacts

Validation 2: YAML ↔ Lineage CSV Consistency

What to Check

Every Gold column in the YAML must have a corresponding entry in COLUMN_LINEAGE.csv.

Validation Pattern

import csv

def validate_yaml_lineage_consistency(yaml_dir: Path, lineage_csv_path: Path) -> dict:
    """Cross-check YAML columns against lineage CSV entries."""
    
    # Parse YAML schemas
    yaml_columns = set()
    for yaml_file in yaml_dir.rglob("*.yaml"):
        with open(yaml_file) as f:
            spec = yaml.safe_load(f)
        table_name = spec.get("table_name", yaml_file.stem)
        for col in spec.get("columns", []):
            yaml_columns.add(f"{table_name}.{col['name']}")
    
    # Parse lineage CSV
    lineage_columns = set()
    with open(lineage_csv_path) as f:
        reader = csv.DictReader(f)
        for row in reader:
            gold_table = row.get("gold_table", "")
            gold_column = row.get("gold_column", "")
            if gold_table and gold_column:
                lineage_columns.add(f"{gold_table}.{gold_column}")
    
    # Cross-check
    missing_lineage = yaml_columns - lineage_columns
    extra_lineage = lineage_columns - yaml_columns
    
    return {
        "valid": len(missing_lineage) == 0,
        "missing_lineage": sorted(missing_lineage),
        "extra_lineage": sorted(extra_lineage),
        "yaml_column_count": len(yaml_columns),
        "lineage_column_count": len(lineage_columns)
    }

Why This Matters

Columns without lineage entries become implementation ambiguity — the merge script author won't know where the data comes from, leading to guesses and bugs. 33% of implementation bugs trace back to incomplete lineage documentation.

Validation 3: PK/FK Reference Consistency

What to Check

Every FOREIGN KEY in a YAML schema must reference a valid PRIMARY KEY column in the referenced table's YAML schema.

Validation Pattern

def validate_pk_fk_consistency(yaml_dir: Path) -> dict:
    """Validate PK/FK references across all YAML schemas."""
    
    # Parse all YAML schemas
    tables = {}
    for yaml_file in yaml_dir.rglob("*.yaml"):
        with open(yaml_file) as f:
            spec = yaml.safe_load(f)
        table_name = spec.get("table_name", yaml_file.stem)
        
        pk_columns = []
        if "primary_key" in spec:
            pk_columns = spec["primary_key"].get("columns", [])
        
        fk_constraints = spec.get("foreign_keys", [])
        
        tables[table_name] = {
            "pk_columns": set(pk_columns),
            "fk_constraints": fk_constraints,
            "all_columns": {col["name"] for col in spec.get("columns", [])}
        }
    
    issues = []
    
    for table_name, info in tables.items():
        for fk in info["fk_constraints"]:
            ref_table = fk.get("references_table", "")
            ref_column = fk.get("references_column", "")
            fk_column = fk.get("column", "")
            
            # Check FK column exists in this table
            if fk_column not in info["all_columns"]:
                issues.append(
                    f"FK column '{table_name}.{fk_column}' not found in table columns"
                )
            
            # Check referenced table exists
            if ref_table not in tables:
                issues.append(
                    f"FK in '{table_name}' references non-existent table '{ref_table}'"
                )
                continue
            
            # Check referenced column is a PK in the target table
            if ref_column not in tables[ref_table]["pk_columns"]:
                issues.append(
                    f"FK '{table_name}.{fk_column}' → '{ref_table}.{ref_column}' "
                    f"but '{ref_column}' is not a PK in '{ref_table}'"
                )
    
    return {
        "valid": len(issues) == 0,
        "issues": issues,
        "table_count": len(tables),
        "total_fk_count": sum(len(t["fk_constraints"]) for t in tables.values())
    }

Common FK Issues

Issue	Cause	Fix
FK references non-existent table	Table renamed or not yet designed	Add missing table or fix reference
FK column not a PK in target	Wrong column referenced	Update FK to reference correct PK
FK column missing from source table	Column removed during design iteration	Add column back or remove FK

Validation 4: YAML Mandatory Fields

What to Check

Every YAML schema must include the non-negotiable defaults from the design orchestrator.

Validation Pattern

def validate_yaml_mandatory_fields(yaml_dir: Path) -> dict:
    """Validate all YAML schemas include mandatory fields."""
    
    MANDATORY_TABLE_PROPERTIES = {
        "delta.enableChangeDataFeed": "true",
        "delta.enableRowTracking": "true",
        "delta.autoOptimize.optimizeWrite": "true",
        "delta.autoOptimize.autoCompact": "true",
        "layer": "gold"
    }
    
    issues = []
    
    for yaml_file in yaml_dir.rglob("*.yaml"):
        with open(yaml_file) as f:
            spec = yaml.safe_load(f)
        table_name = spec.get("table_name", yaml_file.stem)
        
        # Check clustering
        if spec.get("clustering") != "auto":
            issues.append(f"{table_name}: missing 'clustering: auto'")
        
        # Check table properties
        props = spec.get("table_properties", {})
        for prop, expected_value in MANDATORY_TABLE_PROPERTIES.items():
            if props.get(prop) != expected_value:
                issues.append(
                    f"{table_name}: missing or wrong table property "
                    f"'{prop}' (expected '{expected_value}', "
                    f"got '{props.get(prop, 'MISSING')}')"
                )
        
        # Check PK columns are NOT NULL
        pk_columns = set()
        if "primary_key" in spec:
            pk_columns = set(spec["primary_key"].get("columns", []))
        
        for col in spec.get("columns", []):
            if col["name"] in pk_columns and col.get("nullable", True):
                issues.append(
                    f"{table_name}.{col['name']}: PK column is nullable "
                    f"(must be 'nullable: false')"
                )
    
    return {
        "valid": len(issues) == 0,
        "issues": issues
    }

Validation 5: Cross-Worker Semantic Rule Compliance

What to Check

Structural validations 1-4 only confirm that fields exist and references resolve. They pass even when the design has used the wrong transformation type, paraphrased the FK shape, left true/false in business dimensions, or forgot an unknown-member row. Validation 5 closes this gap by enforcing the semantic rules shared across the design-worker skills.

Rule Table

#	Rule	Authoritative source	How to detect a violation
1	No BOOLEAN columns in business-sourced dimensions	design-workers/02-dimension-patterns Rule 3	col.type == "BOOLEAN" in any dim_*.yaml except dim_date / dim_time whitelist
2	Every dimension referenced by a nullable: true FK declares unknown_member:	references/yaml-schema-patterns.md Unknown Member section	Scan fact YAMLs for nullable: true FKs, then check target dim has an unknown_member block
3	Every lineage.transformation value is in the 15-item enum	00-gold-layer-design/SKILL.md Phase 4	transformation not in STANDARD_TRANSFORMATIONS
4	Every foreign_keys: entry uses {columns, references, nullable} shape	DESIGN_DECISIONS.md FK contract	Missing any of the three keys, or extra keys
5	No literal [, ], <, or > characters inside description: strings	common/naming-tagging-standards description pattern	re.search(r"[\[\]<>]", description) matches
6	Mandatory top-level YAML keys present (table_name, domain, description, table_properties, clustering, columns, primary_key)	DESIGN_DECISIONS.md top-level key contract	Any mandatory key missing