Data Governance: Classification, Tagging & Data Quality

-- Create a tag for sensitivity levels
CREATE TAG IF NOT EXISTS sensitivity_level
  ALLOWED_VALUES 'high', 'medium', 'low';

-- Apply tag to a column
ALTER TABLE orders MODIFY COLUMN customer_email SET TAG sensitivity_level = 'high';

-- Query tag references
SELECT * FROM TABLE(INFORMATION_SCHEMA.TAG_REFERENCES('orders', 'TABLE'));

-- Use system function
SELECT SYSTEM$TAG_REFERENCE('orders', 'TABLE', 'sensitivity_level');

CREATE OR REPLACE PROCEDURE classify_columns()
RETURNS VARCHAR
LANGUAGE JAVASCRIPT
AS
$$
  var result = '';
  var tables = snowflake.execute({
    sqlText: `SELECT TABLE_CATALOG, TABLE_SCHEMA, TABLE_NAME, COLUMN_NAME, DATA_TYPE
              FROM INFORMATION_SCHEMA.COLUMNS
              WHERE TABLE_SCHEMA = 'PUBLIC'`
  });
  while (tables.next()) {
    var catalog = tables.getColumnValue(1);
    var schema = tables.getColumnValue(2);
    var table = tables.getColumnValue(3);
    var column = tables.getColumnValue(4);
    var type = tables.getColumnValue(5);
    var tagValue = 'low';
    if (column.match(/email|ssn|phone|credit/i)) {
      tagValue = 'high';
    } else if (type === 'VARCHAR' && column.match(/name|address/i)) {
      tagValue = 'medium';
    }
    var alterSQL = `ALTER TABLE ${catalog}.${schema}.${table} MODIFY COLUMN ${column} SET TAG sensitivity_level = '${tagValue}'`;
    try {
      snowflake.execute({sqlText: alterSQL});
      result += 'Tagged ' + column + ' as ' + tagValue + '\\n';
    } catch(err) {
      result += 'Error tagging ' + column + ': ' + err + '\\n';
    }
  }
  return result;
$$;

-- Schedule with a task
CREATE OR REPLACE TASK classify_task
  WAREHOUSE = my_wh
  SCHEDULE = 'USING CRON 0 6 * * * UTC'
AS
  CALL classify_columns();

ALTER TASK classify_task RESUME;

-- Create masking policy that masks based on role
CREATE OR REPLACE MASKING POLICY mask_high AS (val VARCHAR) RETURNS VARCHAR ->
  CASE
    WHEN CURRENT_ROLE() IN ('ADMIN', 'COMPLIANCE') THEN val
    ELSE '***MASKED***'
  END;

-- Attach policy to tag
ALTER TAG sensitivity_level SET MASKING POLICY mask_high;

-- Now any column with tag sensitivity_level = 'high' will be masked for non-privileged roles.
-- Test as analyst role
USE ROLE ANALYST;
SELECT customer_email FROM orders; -- returns '***MASKED***'

-- Create a table to store quality metrics
CREATE TABLE quality_metrics (
  table_name VARCHAR,
  column_name VARCHAR,
  null_count NUMBER,
  total_rows NUMBER,
  null_ratio FLOAT,
  checked_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP()
);

-- Task to compute metrics daily
CREATE OR REPLACE TASK compute_quality
  WAREHOUSE = my_wh
  SCHEDULE = 'USING CRON 0 7 * * * UTC'
AS
  INSERT INTO quality_metrics (table_name, column_name, null_count, total_rows, null_ratio)
  SELECT 'orders', 'order_amount',
         COUNT(*)-COUNT(order_amount),
         COUNT(*),
         (COUNT(*)-COUNT(order_amount))/NULLIF(COUNT(*),0)
  FROM orders;

-- Alert if null ratio > 0.01
CREATE OR REPLACE ALERT high_null_rate
  WAREHOUSE = my_wh
  SCHEDULE = 'USING CRON 0 8 * * * UTC'
  IF (EXISTS (
    SELECT 1 FROM quality_metrics
    WHERE null_ratio > 0.01 AND checked_at >= CURRENT_TIMESTAMP() - INTERVAL '1 day'
  ))
  THEN
    CALL SYSTEM$SEND_EMAIL('governance@company.com', 'High NULL rate detected', 'Check quality_metrics table.');

ALTER TASK compute_quality RESUME;
ALTER ALERT high_null_rate RESUME;

-- Find all columns without a specific tag
SELECT TABLE_CATALOG, TABLE_SCHEMA, TABLE_NAME, COLUMN_NAME
FROM INFORMATION_SCHEMA.COLUMNS
WHERE (TABLE_CATALOG, TABLE_SCHEMA, TABLE_NAME, COLUMN_NAME) NOT IN (
  SELECT OBJECT_DATABASE, OBJECT_SCHEMA, OBJECT_NAME, COLUMN_NAME
  FROM TABLE(INFORMATION_SCHEMA.TAG_REFERENCES_ALL_COLUMNS())
  WHERE TAG_NAME = 'SENSITIVITY_LEVEL'
)
AND TABLE_SCHEMA = 'PUBLIC';

-- List all tags applied to columns
SELECT * FROM TABLE(INFORMATION_SCHEMA.TAG_REFERENCES_ALL_COLUMNS())
WHERE TAG_NAME = 'SENSITIVITY_LEVEL';

-- Assume a mapping table that links tag values to roles
CREATE TABLE region_access (
  region_tag VARCHAR,
  allowed_role VARCHAR
);
INSERT INTO region_access VALUES ('US', 'ANALYST_US'), ('EU', 'ANALYST_EU');

-- Row access policy that filters based on current role
CREATE OR REPLACE ROW ACCESS POLICY region_filter AS (region_tag VARCHAR) RETURNS BOOLEAN ->
  EXISTS (
    SELECT 1 FROM region_access
    WHERE region_tag = region_tag
      AND allowed_role = CURRENT_ROLE()
  );

-- Apply policy to table (assuming table has a column 'region' tagged with 'region_tag')
ALTER TABLE sales ADD ROW ACCESS POLICY region_filter ON (region);

-- Now users only see rows where their role matches the region tag.

-- Example: Create a governance dashboard view
CREATE OR REPLACE VIEW governance_dashboard AS
SELECT 
  t.TABLE_CATALOG,
  t.TABLE_SCHEMA,
  t.TABLE_NAME,
  COUNT(DISTINCT c.COLUMN_NAME) AS total_columns,
  COUNT(DISTINCT tr.COLUMN_NAME) AS tagged_columns,
  COUNT(DISTINCT c.COLUMN_NAME) - COUNT(DISTINCT tr.COLUMN_NAME) AS untagged_columns
FROM INFORMATION_SCHEMA.TABLES t
JOIN INFORMATION_SCHEMA.COLUMNS c ON t.TABLE_CATALOG = c.TABLE_CATALOG AND t.TABLE_SCHEMA = c.TABLE_SCHEMA AND t.TABLE_NAME = c.TABLE_NAME
LEFT JOIN TABLE(INFORMATION_SCHEMA.TAG_REFERENCES_ALL_COLUMNS()) tr 
  ON c.TABLE_CATALOG = tr.OBJECT_DATABASE AND c.TABLE_SCHEMA = tr.OBJECT_SCHEMA AND c.TABLE_NAME = tr.OBJECT_NAME AND c.COLUMN_NAME = tr.COLUMN_NAME
WHERE t.TABLE_SCHEMA = 'PUBLIC'
GROUP BY t.TABLE_CATALOG, t.TABLE_SCHEMA, t.TABLE_NAME;