Evaluate Data Quality in Snowflake

SELECT comment
FROM <my_database>.INFORMATION_SCHEMA.FUNCTIONS
WHERE function_name = 'ATC_DQ_EMAIL_VALIDATION';

deployed_at: 2025-08-28 09:09:00;
firewall_id: B5MSbQjSJr;
firewall_filename: email_validation.zip;
firewall_download_time: 2025-08-28 09:08:05;
arguments: #1: surname - STRING, #2: email - STRING;
returns: OBJECT<passed: boolean, score: int, rule_instance_results: OBJECT>;

{
  "passed": false,
  "rule_instance_results": [
    {
      "dimension_result": "INVALID",
      "explanation": "FUTURE_DATE",
      "overall_result": "FAILED",
      "rule_instance_id": "Date is not in future"
    },
    {
      "dimension_result": "VALID",
      "explanation": "OTHER",
      "overall_result": "PASSED",
      "rule_instance_id": "Date is in fiscal season range"
    }
  ],
  "score": 2000
}

-- Insert only valid records
INSERT INTO clean_customers
SELECT firstname, lastname, email, phone
FROM raw_customers
WHERE MY_DB.MY_SCHEMA.atc_dq_customers(firstname, lastname):passed = TRUE;

-- Auto-ingest with validation
CREATE PIPE customer_validation_pipe AS
COPY INTO validated_customers
FROM (
  SELECT
    $1:firstname::STRING as firstname,
    $1:lastname::STRING as lastname,
    MY_DB.MY_SCHEMA.atc_dq_customers($1:firstname::STRING, $1:lastname::STRING) as quality_result
  FROM @my_stage
  WHERE MY_DB.MY_SCHEMA.atc_dq_customers($1:firstname::STRING, $1:lastname::STRING):passed = TRUE
);

CREATE OR REPLACE TABLE customers_with_quality AS
SELECT
  *,
  MY_DB.MY_SCHEMA.atc_dq_customers(firstname, lastname) as quality_check,
  MY_DB.MY_SCHEMA.atc_dq_email(email) as email_quality
FROM raw_customers;

-- Check data quality before proceeding
SET quality_pass_rate = (
  SELECT
    COUNT_IF(MY_DB.MY_SCHEMA.atc_dq_customers(firstname, lastname):passed) * 100.0 / COUNT(*)
  FROM raw_customers
);

-- Conditional processing based on quality
IF ($quality_pass_rate < 95) THEN
  CALL SYSTEM$SEND_EMAIL('dq-alerts@company.com', 'Data Quality Alert', 'Quality below threshold');
  RETURN 'PIPELINE_FAILED';
END IF;

CREATE OR REPLACE VIEW daily_quality_report AS
SELECT
  DATE(created_at) as report_date,
  COUNT(*) as total_records,
  COUNT_IF(MY_DB.MY_SCHEMA.atc_dq_customers(firstname, lastname):passed) as valid_records,
  COUNT_IF(NOT MY_DB.MY_SCHEMA.atc_dq_customers(firstname, lastname):passed) as invalid_records
FROM customers
GROUP BY DATE(created_at);

{% test validate_customers(model, first_name_column, last_name_column) %}
select *
from {{ model }}
where MY_DB.MY_SCHEMA.atc_dq_customers({{ first_name_column }}, {{ last_name_column }}):passed = FALSE
{% endtest %}