03_dq_rules_engine.py

# Fabric Notebook: 03 — DQ Rules Engine (Core)
# ==============================================
# The generic, metadata-driven data quality rules engine.
#
# This notebook:
#   1. Loads target data, reference data, and rules config
#   2. Joins target ↔ reference to enrich records
#   3. Matches enriched records to applicable rules (multi-criteria + wildcards)
#   4. Evaluates each rule expression dynamically via Spark SQL
#   5. Writes detailed pass/fail results to dq_results Delta table
#
# This notebook contains NO business logic. All rules are in dq_rules_config.
# ==============================================

from pyspark.sql.functions import (
    col, lit, current_timestamp, expr, when, 
    monotonically_increasing_id, concat_ws, coalesce
)
from pyspark.sql.types import StringType, BooleanType
import uuid
from datetime import datetime

# -------------------------------------------------------
# PARAMETERS — Change these to point to your tables
# -------------------------------------------------------

TARGET_TABLE      = "claims"
REFERENCE_TABLE   = "policy_criteria"
JOIN_KEY          = "policy_id"
RULES_TABLE       = "dq_rules_config"
RESULTS_TABLE     = "dq_results"

# Criteria columns used for rule matching
# These must exist in both the enriched data and the rules config table
CRITERIA_COLUMNS  = ["policy_type", "region", "product", "risk_tier", "channel"]

# -------------------------------------------------------
# 1. Generate run metadata
# -------------------------------------------------------

run_id = str(uuid.uuid4())
run_timestamp = datetime.utcnow()
print(f"DQ Run ID: {run_id}")
print(f"Run Timestamp: {run_timestamp}")

# -------------------------------------------------------
# 2. Load data
# -------------------------------------------------------

print("\n--- Loading data ---")
target_df = spark.read.format("delta").table(TARGET_TABLE)
print(f"  Target table '{TARGET_TABLE}': {target_df.count()} records")

reference_df = spark.read.format("delta").table(REFERENCE_TABLE)
print(f"  Reference table '{REFERENCE_TABLE}': {reference_df.count()} records")

rules_df = spark.read.format("delta").table(RULES_TABLE)
print(f"  Rules table '{RULES_TABLE}': {rules_df.count()} total rules")

# -------------------------------------------------------
# 3. Filter active rules
# -------------------------------------------------------

from pyspark.sql.functions import current_date

active_rules = rules_df.filter(
    (col("is_active") == True) &
    (col("effective_from") <= current_date()) &
    ((col("effective_to").isNull()) | (col("effective_to") >= current_date()))
)

active_rule_count = active_rules.count()
print(f"  Active rules: {active_rule_count}")

if active_rule_count == 0:
    print("⚠ No active rules found. Exiting.")
    raise SystemExit("No active rules to evaluate")

# -------------------------------------------------------
# 4. Enrich target data (join with reference)
# -------------------------------------------------------

print("\n--- Enriching target data ---")
enriched_df = target_df.join(
    reference_df,
    on=JOIN_KEY,
    how="left"
)

enriched_count = enriched_df.count()
print(f"  Enriched records: {enriched_count}")

# Check for unmatched records (no reference data)
unmatched = enriched_df.filter(col("policy_type").isNull()).count()
if unmatched > 0:
    print(f"  ⚠ {unmatched} records have no matching reference data — rules requiring criteria will not match")

# -------------------------------------------------------
# 5. Match records to rules and evaluate
# -------------------------------------------------------

print("\n--- Matching and evaluating rules ---")

# Collect active rules to driver for iteration
# (Rules table is always small — safe to collect)
active_rules_list = active_rules.collect()

all_results = []

for rule_row in active_rules_list:
    rule_id = rule_row["rule_id"]
    rule_name = rule_row["rule_name"]
    severity = rule_row["severity"]
    target_column = rule_row["target_column"]
    rule_expression = rule_row["rule_expression"]
    
    # Build match condition with wildcard support
    # For each criteria column: rule value is * OR rule value matches record value
    match_conditions = []
    for criteria_col in CRITERIA_COLUMNS:
        rule_value = rule_row[criteria_col]
        if rule_value == "*":
            # Wildcard — matches everything, no filter needed
            continue
        else:
            # Exact match required
            match_conditions.append(col(criteria_col) == lit(rule_value))
    
    # Start with all enriched records
    matched_records = enriched_df
    
    # Apply match conditions
    for condition in match_conditions:
        matched_records = matched_records.filter(condition)
    
    matched_count = matched_records.count()
    
    if matched_count == 0:
        print(f"  Rule {rule_id} ({rule_name}): 0 matching records — skipped")
        continue
    
    # Evaluate the rule expression
    try:
        evaluated = matched_records.withColumn(
            "rule_passed",
            expr(rule_expression).cast(BooleanType())
        )
    except Exception as e:
        print(f"  ✗ Rule {rule_id} ({rule_name}): Expression error — {str(e)}")
        # Create results marking all as failed due to expression error
        evaluated = matched_records.withColumn("rule_passed", lit(None).cast(BooleanType()))
    
    # Build result rows
    result_df = evaluated.select(
        lit(run_id).alias("run_id"),
        lit(str(run_timestamp)).cast("timestamp").alias("run_timestamp"),
        col("claim_id"),
        col("policy_id"),
        lit(rule_id).alias("rule_id"),
        lit(rule_name).alias("rule_name"),
        lit(severity).alias("severity"),
        lit(target_column).alias("target_column"),
        lit(rule_expression).alias("rule_expression"),
        col("rule_passed"),
        # Capture the actual value of the target column for traceability
        coalesce(col(target_column).cast(StringType()), lit("NULL")).alias("actual_value"),
        # Include criteria context
        coalesce(col("policy_type"), lit("N/A")).alias("policy_type"),
        coalesce(col("region"), lit("N/A")).alias("region"),
        coalesce(col("product"), lit("N/A")).alias("product"),
        coalesce(col("risk_tier"), lit("N/A")).alias("risk_tier"),
        coalesce(col("channel"), lit("N/A")).alias("channel"),
        current_timestamp().alias("evaluated_at"),
    )
    
    passed = evaluated.filter(col("rule_passed") == True).count()
    failed = evaluated.filter(col("rule_passed") == False).count()
    null_results = evaluated.filter(col("rule_passed").isNull()).count()
    
    status_icon = "✓" if failed == 0 else "✗"
    print(f"  {status_icon} Rule {rule_id} ({rule_name}): {matched_count} matched, {passed} passed, {failed} failed" + 
          (f", {null_results} null/error" if null_results > 0 else ""))
    
    all_results.append(result_df)

# -------------------------------------------------------
# 6. Union all results and write
# -------------------------------------------------------

if len(all_results) == 0:
    print("\n⚠ No rule evaluations produced results. Nothing to write.")
else:
    from functools import reduce
    from pyspark.sql import DataFrame
    
    final_results = reduce(DataFrame.unionByName, all_results)
    
    # Add a unique result ID
    final_results = final_results.withColumn(
        "result_id",
        concat_ws("-", lit(run_id), monotonically_increasing_id().cast(StringType()))
    )
    
    # Write to Delta table (append mode — preserves history)
    final_results.write.format("delta").mode("append").saveAsTable(RESULTS_TABLE)
    
    total_results = final_results.count()
    total_passed = final_results.filter(col("rule_passed") == True).count()
    total_failed = final_results.filter(col("rule_passed") == False).count()
    
    print(f"\n--- Run Summary ---")
    print(f"  Run ID:         {run_id}")
    print(f"  Total evaluations: {total_results}")
    print(f"  Passed:         {total_passed}")
    print(f"  Failed:         {total_failed}")
    print(f"  Pass rate:      {total_passed/total_results*100:.1f}%")
    print(f"  Results written to: {RESULTS_TABLE}")

# -------------------------------------------------------
# 7. Preview failures
# -------------------------------------------------------

print("\n--- Failed Rules Detail ---")
spark.sql(f"""
    SELECT claim_id, rule_id, rule_name, severity, target_column, 
           rule_expression, actual_value, policy_type, region, product
    FROM {RESULTS_TABLE}
    WHERE run_id = '{run_id}' AND rule_passed = false
    ORDER BY severity, claim_id, rule_id
""").show(50, truncate=False)

print("\n✓ DQ Rules Engine execution complete. Proceed to notebook 04 for summary report.")