-- Check for null values
SELECT
  'email' as column_name,
  COUNT(*) as total_rows,
  COUNT(email) as non_null_rows,
  ROUND(COUNT(email) / COUNT(*) * 100, 2) as completeness_pct
FROM customers

UNION ALL

SELECT
  'phone' as column_name,
  COUNT(*) as total_rows,
  COUNT(phone) as non_null_rows,
  ROUND(COUNT(phone) / COUNT(*) * 100, 2) as completeness_pct
FROM customers;

-- Create completeness validation
CREATE OR REPLACE PROCEDURE check_completeness(
  table_name STRING,
  column_name STRING,
  threshold FLOAT
)
RETURNS STRING
LANGUAGE SQL
AS
$$
DECLARE
  completeness FLOAT;
BEGIN
  EXECUTE IMMEDIATE
    'SELECT COUNT(' || column_name || ') / COUNT(*) FROM ' || table_name
    INTO completeness;
  IF completeness >= threshold THEN
    RETURN 'PASS: ' || ROUND(completeness * 100, 2) || '% completeness';
  ELSE
    RETURN 'FAIL: ' || ROUND(completeness * 100, 2) || '% completeness (threshold: ' || ROUND(threshold * 100, 2) || '%)';
  END IF;
END;
$$;

-- Execute completeness check
CALL check_completeness('CUSTOMERS', 'EMAIL', 0.95);

-- Validate email format
-- Create accuracy validation
DECLARE
  invalid_count INTEGER;
BEGIN
  -- Check invalid emails
  SELECT COUNT(*) INTO invalid_count
  FROM customers
  WHERE NOT REGEXP_LIKE(email, '^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$');

  IF invalid_count > 0 THEN
    RETURN 'ERROR: ' || invalid_count || ' invalid emails found';
  END IF;

  -- Check invalid dates
  SELECT COUNT(*) INTO invalid_count
  FROM orders
  WHERE order_date > CURRENT_DATE() OR order_date < '2000-01-01';

  IF invalid_count > 0 THEN
    RETURN 'ERROR: ' || invalid_count || ' invalid dates found';
  END IF;

  RETURN 'SUCCESS: All data accurate';
END;
$$;

-- Check cross-table consistency
SELECT
  o.customer_id,
  COUNT(o.order_id) as order_count,
  CASE
    WHEN c.customer_id IS NULL THEN 'ORPHAN_ORDER'
    ELSE 'CONSISTENT'
  END as consistency_status
FROM orders o
LEFT JOIN customers c ON o.customer_id = c.customer_id
WHERE c.customer_id IS NULL;

-- Create consistency validation
CREATE OR REPLACE PROCEDURE check_cross_table_consistency()
RETURNS STRING
LANGUAGE SQL
AS
$$
DECLARE
  orphan_count INTEGER;
BEGIN
  SELECT COUNT(*) INTO orphan_count
  FROM orders o
  LEFT JOIN customers c ON o.customer_id = c.customer_id
  WHERE c.customer_id IS NULL;
  IF orphan_count > 0 THEN
    RETURN 'ERROR: ' || orphan_count || ' orphan orders found';
  END IF;
  RETURN 'SUCCESS: Cross-table consistency maintained';
END;
$$;

-- Comprehensive data profiling
DECLARE
  result VARIANT;
BEGIN
  EXECUTE IMMEDIATE
    'SELECT
      COUNT(*) as total_rows,
      COUNT(DISTINCT *) as unique_values,
      MIN(created_at) as min_date,
      MAX(created_at) as max_date,
      AVG(amount) as avg_amount,
      STDDEV(amount) as stddev_amount
    FROM ' || table_name
    INTO result;

  RETURN result;
END;
$$;

-- Profile specific columns
SELECT
  COUNT(*) as total_rows,
  COUNT(DISTINCT customer_id) as unique_customers,
  MIN(order_date) as min_date,
  MAX(order_date) as max_date,
  AVG(amount) as avg_order_amount,
  STDDEV(amount) as stddev_amount,
  PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY amount) as median_amount
FROM orders;

-- Create quality rules table
CREATE OR REPLACE TABLE data_quality_rules (
  rule_id INTEGER PRIMARY KEY,
  table_name STRING,
  column_name STRING,
  rule_type STRING,
  rule_expression STRING,
  threshold FLOAT,
  enabled BOOLEAN DEFAULT TRUE
);

-- Insert quality rules
INSERT INTO data_quality_rules VALUES
  (1, 'CUSTOMERS', 'EMAIL', 'NOT_NULL', 'email IS NOT NULL', 0.95, TRUE),
  (2, 'CUSTOMERS', 'EMAIL', 'FORMAT', 'REGEXP_LIKE(email, ''^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$'')', 0.98, TRUE),
  (3, 'ORDERS', 'AMOUNT', 'RANGE', 'amount > 0 AND amount < 1000000', 0.99, TRUE),
  (4, 'ORDERS', 'ORDER_DATE', 'NOT_FUTURE', 'order_date <= CURRENT_DATE()', 1.0, TRUE);

-- Execute quality rules
CREATE OR REPLACE PROCEDURE execute_quality_rules()
RETURNS STRING
LANGUAGE SQL
AS
$$
DECLARE
  rule_record RECORD;
  total_rows INTEGER;
  passing_rows INTEGER;
  result STRING;
BEGIN
  FOR rule_record IN
    SELECT * FROM data_quality_rules WHERE enabled = TRUE
  DO
    -- Count total rows
    EXECUTE IMMEDIATE
      'SELECT COUNT(*) FROM ' || rule_record.table_name
      INTO total_rows;
    -- Count passing rows
    EXECUTE IMMEDIATE
      'SELECT COUNT(*) FROM ' || rule_record.table_name
      || ' WHERE ' || rule_record.rule_expression
      INTO passing_rows;
    -- Check threshold
    IF (passing_rows::FLOAT / total_rows) < rule_record.threshold THEN
      result := result || 'FAIL: Rule ' || rule_record.rule_id || ' - ' ||
        ROUND((passing_rows::FLOAT / total_rows) * 100, 2) || '% passing\n';
    END IF;
  END FOR;
  RETURN COALESCE(result, 'SUCCESS: All rules passing');
END;
$$;
-- Schedule rule execution
CREATE TASK daily_quality_check
SCHEDULE = 'USING CRON 0 2 * * * America/New_York'
AS
  CALL execute_quality_rules();

-- Auto-fix common issues
DECLARE
  fixed_count INTEGER;
BEGIN
  -- Fix invalid emails
  UPDATE customers
  SET email = LOWER(TRIM(email))
  WHERE email != LOWER(TRIM(email));

  GET DIAGNOSTICS fixed_count = ROW_COUNT;

  -- Fix future dates
  UPDATE orders
  SET order_date = CURRENT_DATE()
  WHERE order_date > CURRENT_DATE();

  GET DIAGNOSTICS fixed_count = fixed_count + ROW_COUNT;

  RETURN 'Fixed ' || fixed_count || ' records';
END;
$$;

-- Create data quality exception table
CREATE TABLE data_quality_exceptions (
  exception_id INTEGER PRIMARY KEY,
  table_name STRING,
  column_name STRING,
  rule_id INTEGER,
  exception_value STRING,
  detected_at TIMESTAMP_NTZ,
  resolved_at TIMESTAMP_NTZ,
  resolution_notes STRING
);