| 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."""
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", [])
}
erd_tables = {}
with open(erd_path) as f:
erd_content = f.read()
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])
erd_tables[table_name] = columns
issues = []
for table in yaml_tables:
if table not in erd_tables:
issues.append(f"YAML table '{table}' missing from ERD")
for table in erd_tables:
if table not in yaml_tables:
issues.append(f"ERD table '{table}' missing from YAML")
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."""
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']}")
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}")
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."""
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", "")
if fk_column not in info["all_columns"]:
issues.append(
f"FK column '{table_name}.{fk_column}' not found in table columns"
)
if ref_table not in tables:
issues.append(
f"FK in '{table_name}' references non-existent table '{ref_table}'"
)
continue
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)
if spec.get("clustering") != "auto":
issues.append(f"{table_name}: missing 'clustering: auto'")
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')}')"
)
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 |
| 7 | Fact measures declare `additivity: additive | semi_additive | non_additive` |
Validation Pattern
import re
import yaml
from pathlib import Path
STANDARD_TRANSFORMATIONS = {
"DIRECT_COPY", "RENAME", "CAST",
"AGGREGATE_SUM", "AGGREGATE_SUM_CONDITIONAL",
"AGGREGATE_COUNT", "AGGREGATE_AVG",
"DERIVED_CALCULATION", "DERIVED_CONDITIONAL",
"HASH_MD5", "HASH_SHA256",
"COALESCE", "DATE_TRUNC", "GENERATED", "LOOKUP",
}
MANDATORY_FK_KEYS = {"columns", "references", "nullable"}
DESC_FORBIDDEN_CHARS = re.compile(r"[\[\]<>]")
BOOLEAN_DIM_WHITELIST = {"dim_date", "dim_time"}
def validate_semantic_rules(yaml_dir: Path) -> dict:
"""Cross-worker semantic rule compliance (Validation 5)."""
issues = []
dim_unknown_members = {}
fact_nullable_fks = []
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)
is_dim = table_name.startswith("dim_")
is_fact = table_name.startswith("fact_")
for key in ("table_name", "domain", "description", "table_properties", "clustering", "columns", "primary_key"):
if key not in spec:
issues.append(f"{table_name}: missing mandatory top-level key '{key}'")
if is_dim and table_name not in BOOLEAN_DIM_WHITELIST:
for col in spec.get("columns", []):
if col.get("type") == "BOOLEAN":
issues.append(
f"{table_name}.{col['name']}: BOOLEAN not allowed in business dimension "
f"(convert to STRING per 02-dimension-patterns Rule 3)"
)
if is_dim:
dim_unknown_members[table_name] = "unknown_member" in spec
for fk in spec.get("foreign_keys", []):
keys = set(fk.keys())
missing = MANDATORY_FK_KEYS - keys
if missing:
issues.append(
f"{table_name}: foreign_keys entry missing keys {sorted(missing)} "
f"(got {sorted(keys)})"
)
if is_fact and fk.get("nullable") is True:
ref = fk.get("references", "")
target = ref.split("(")[0].strip()
if target:
fact_nullable_fks.append((table_name, target))
for col in spec.get("columns", []):
desc = col.get("description", "")
if DESC_FORBIDDEN_CHARS.search(desc):
issues.append(
f"{table_name}.{col['name']}: description contains literal bracket/angle char "
f"(placeholders are not literal per naming-tagging-standards)"
)
lineage = col.get("lineage") or {}
tfm = lineage.get("transformation")
if tfm and tfm not in STANDARD_TRANSFORMATIONS:
issues.append(
f"{table_name}.{col['name']}: transformation '{tfm}' not in 15-type enum "
f"(see 00-gold-layer-design Phase 4 edge-case mapping)"
)
if is_fact:
for m in spec.get("measures", []):
if "additivity" not in m:
issues.append(
f"{table_name}: measure '{m.get('name', '?')}' missing 'additivity' "
f"(additive | semi_additive | non_additive)"
)
for fact_name, target_dim in fact_nullable_fks:
if target_dim in dim_unknown_members and not dim_unknown_members[target_dim]:
issues.append(
f"{target_dim}: referenced by {fact_name} with nullable: true FK, "
f"but dimension does not declare 'unknown_member:' block"
)
return {"valid": len(issues) == 0, "issues": issues}
Wiring Into the Full Workflow
Add Validation 5 alongside the existing four in run_design_validation:
results["semantic_rules"] = validate_semantic_rules(yaml_dir)
Validation 6: Industry Data Model Alignment (ADVISORY)
What to Check
Benchmark the Gold design against a Databricks Industry Vibe Data Model (or any canonical industry reference) for entity coverage, terminology, and sensitivity alignment. Unlike Validations 1–5 — self-referential consistency checks that MUST pass — Validation 6 is advisory: it produces a coverage scorecard and gap list but NEVER flips the overall all_valid result.
Read first: design-workers/08-industry-alignment/references/industry-data-model-alignment.md — the Silver-vs-Gold nuance, the manifest, the crosswalk states, and the scoring rubric. This is semantic coverage, NOT structural equivalence.
Single Source of Matching
Matching is owned by design-workers/08-industry-alignment (Phase 0/2), which writes gold_layer_design/INDUSTRY_CROSSWALK.csv. Validation 6 loads that crosswalk — it does NOT re-match — then verifies it against the current YAML and scores it. This avoids two divergent matchers.
Preconditions
- If
INDUSTRY_CROSSWALK.csv is absent, industry_reference_source: none (or the worker was skipped). Record "Industry alignment: N/A" and return {"valid": True, "applicable": False} — never a failure.
Rule Table
| # | Rule | Severity | How detected |
|---|
| 1 | Every core reference entity is Covered, Absorbed, Waived, or Planned | advisory-high | Crosswalk row in Gap state with importance: core |
| 2 | Covered/Absorbed gold_tables still exist in the current YAML | advisory-high | Named Gold table not found among YAML table_names (drift after Phase 2) |
| 3 | Every PII entity's covering Gold table carries a PII tag | advisory-med | Covering table missing table_properties.PII |
| 4 | Every Waived/Planned/Gap row has a rationale | advisory-low | Empty rationale cell |
Validation Pattern
import csv
import yaml
from pathlib import Path
IMPORTANCE_WEIGHT = {"core": 3, "extended": 2, "optional": 1}
def _load_gold_tables(yaml_dir: Path) -> dict:
tables = {}
for yaml_file in yaml_dir.rglob("*.yaml"):
with open(yaml_file) as f:
spec = yaml.safe_load(f) or {}
name = spec.get("table_name", yaml_file.stem)
props = spec.get("table_properties", {}) or {}
tables[name] = {"has_pii": str(props.get("PII", "")).lower() in {"true", "yes", "pii"}}
return tables
def validate_industry_alignment(yaml_dir: Path, crosswalk_csv: Path) -> dict:
"""Validation 6 — advisory. Loads the worker's crosswalk, verifies it against the
current YAML, and scores coverage / terminology / PII. `valid` is always True."""
if not crosswalk_csv or not Path(crosswalk_csv).exists():
return {"valid": True, "applicable": False,
"note": "INDUSTRY_CROSSWALK.csv absent — industry alignment N/A."}
gold = _load_gold_tables(yaml_dir)
rows = list(csv.DictReader(open(crosswalk_csv)))
issues = []
covered_w = applicable_w = 0
pii_hits = pii_total = 0
for r in rows:
state = r["state"].strip()
importance = r.get("importance", "optional").strip()
weight = IMPORTANCE_WEIGHT.get(importance, 1)
gold_tables = [t for t in (r.get("gold_tables", "") or "").split("|") if t]
sensitive = r.get("sensitivity", "none").strip().upper() == "PII"
if state not in {"Waived", "Planned"}:
applicable_w += weight
if state in {"Covered", "Absorbed"}:
covered_w += weight
if state == "Gap" and importance in {"core", "extended"}:
issues.append(f"[gap] {importance} entity '{r['entity']}' not covered/waived/planned")
for t in gold_tables:
if t not in gold:
issues.append(f"[drift] '{r['entity']}' maps to '{t}' which is not in the current YAML")
if sensitive and state in {"Covered", "Absorbed"}:
pii_total += 1
if any(gold.get(t, {}).get("has_pii") for t in gold_tables):
pii_hits += 1
else:
issues.append(f"[pii] '{r['entity']}' is PII but covering table(s) {gold_tables} lack a PII tag")
if state in {"Waived", "Planned", "Gap"} and not (r.get("rationale", "") or "").strip():
issues.append(f"[rationale] '{r['entity']}' is {state} but has no rationale")
def pct(a, b):
return round(100 * a / b) if b else 100
return {
"valid": True,
"applicable": True,
"scorecard": {
"entity_coverage_pct": pct(covered_w, applicable_w),
"pii_alignment_pct": pct(pii_hits, pii_total),
},
"issues": issues,
}
Emit the Report
From the returned dict, emit gold_layer_design/INDUSTRY_ALIGNMENT.md using the report template in 08-industry-alignment/references/industry-data-model-alignment.md. Record the scorecard line in the validation report; if applicable is False, print "Industry alignment: N/A (no reference model provided)".
Why Advisory, Not a Gate
The reference is external, volatile, and models a different layer (Silver business model vs. Gold dimensional). A hard gate would produce false failures whenever the customer's legitimate scope differs from the industry archetype — contradicting Vibe's own "shaped entirely by YOUR context" philosophy. The value is a documented coverage %, a gap list for Phase 9 stakeholder review, and a terminology/PII sanity check — not a pass/fail wall.
Complete Design Validation Workflow
Run all validations as a comprehensive pre-handoff check (Validations 1–5 are must-pass; Validation 6 is advisory and excluded from all_valid):
def run_design_validation(project_dir: Path) -> dict:
"""Run complete design consistency validation suite."""
yaml_dir = project_dir / "gold_layer_design" / "yaml"
erd_path = project_dir / "gold_layer_design" / "erd_master.md"
lineage_csv = project_dir / "gold_layer_design" / "COLUMN_LINEAGE.csv"
results = {}
if erd_path.exists():
results["yaml_erd"] = validate_yaml_erd_consistency(yaml_dir, erd_path)
else:
results["yaml_erd"] = {"valid": False, "issues": ["ERD file not found"]}
if lineage_csv.exists():
results["yaml_lineage"] = validate_yaml_lineage_consistency(yaml_dir, lineage_csv)
else:
results["yaml_lineage"] = {"valid": False, "issues": ["Lineage CSV not found"]}
results["pk_fk"] = validate_pk_fk_consistency(yaml_dir)
results["mandatory_fields"] = validate_yaml_mandatory_fields(yaml_dir)
results["semantic_rules"] = validate_semantic_rules(yaml_dir)
crosswalk_csv = project_dir / "gold_layer_design" / "INDUSTRY_CROSSWALK.csv"
results["industry_alignment"] = validate_industry_alignment(yaml_dir, crosswalk_csv)
must_pass = {k: v for k, v in results.items() if k != "industry_alignment"}
all_valid = all(r.get("valid", False) for r in must_pass.values())
total_issues = sum(len(r.get("issues", [])) for r in results.values())
print(f"\n{'='*60}")
print(f"Design Consistency Validation Report")
print(f"{'='*60}")
for name, result in results.items():
status = "✅ PASS" if result.get("valid") else "❌ FAIL"
issue_count = len(result.get("issues", []))
print(f" {name}: {status} ({issue_count} issues)")
for issue in result.get("issues", [])[:5]:
print(f" - {issue}")
if issue_count > 5:
print(f" ... and {issue_count - 5} more")
print(f"\nOverall: {'✅ ALL PASS' if all_valid else '❌ ISSUES FOUND'}")
print(f"Total issues: {total_issues}")
return {"all_valid": all_valid, "total_issues": total_issues, "details": results}
Validation Checklist (Design Phase)
Before handing off to implementation:
Reference Files
Related Skills
- Merge Schema Validation (Implementation):
pipeline-workers/05-schema-validation/SKILL.md — Runtime DataFrame-vs-DDL validation
- YAML-Driven Gold Setup:
pipeline-workers/01-yaml-table-setup/SKILL.md — YAML schema structure and format
- Mermaid ERD Patterns:
design-workers/05-erd-diagrams/SKILL.md — ERD syntax and organization
References
Inputs
- From
06-table-documentation: Complete YAML schema files with descriptions, TBLPROPERTIES, and lineage metadata
- From
05-erd-diagrams: ERD diagrams (master, domain, summary) with all tables and relationships
- From orchestrator Phase 5: COLUMN_LINEAGE.csv with Bronze → Silver → Gold mappings
Outputs
- Validation report (pass/fail per category: YAML↔ERD, YAML↔Lineage, PK/FK, mandatory fields)
- List of inconsistencies to fix before implementation handoff
- Completed design sign-off checklist
Design Notes to Carry Forward
After completing this skill, note:
Next Step
Design phase is complete. Return to the orchestrator (00-gold-layer-design) for Phase 9 (Stakeholder Review), then proceed to implementation via gold/01-gold-layer-setup/SKILL.md.
Pattern Origin: Phase 8 of Gold Layer Design workflow, implicit validation patterns made explicit
Key Lesson: Design artifacts drift across phases. Cross-validate before implementation.
Impact: Prevents 33% of implementation bugs caused by incomplete lineage and design inconsistencies