07-design-validation

# 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 ```python 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 ```python 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 ```python 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 ```python 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 |