Governance, Access & Security

SELECT service_name, action_name,
       CASE
         WHEN identity_metadata.run_by IS NULL OR identity_metadata.run_by = '__REDACTED__' THEN identity_metadata.run_by
         WHEN identity_metadata.run_by LIKE '%@%' THEN concat(substr(identity_metadata.run_by, 1, 2), '****@****')
         WHEN identity_metadata.run_by RLIKE '^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$' THEN identity_metadata.run_by
         ELSE concat(substr(identity_metadata.run_by, 1, 2), '****')
       END AS run_by,
       CASE
         WHEN identity_metadata.run_as IS NULL OR identity_metadata.run_as = '__REDACTED__' THEN identity_metadata.run_as
         WHEN identity_metadata.run_as LIKE '%@%' THEN concat(substr(identity_metadata.run_as, 1, 2), '****@****')
         WHEN identity_metadata.run_as RLIKE '^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$' THEN identity_metadata.run_as
         ELSE concat(substr(identity_metadata.run_as, 1, 2), '****')
       END AS run_as,
       COUNT(*) AS event_count,
       MIN(event_time) AS first_event_time, MAX(event_time) AS last_event_time,
       -- status: worst-first band on run-by != run-as event volume per identity pair (field heuristic; :warn_runas_events / :crit_runas_events). Zero rows does not mean OK - see caveats.
       CASE
         WHEN COUNT(*) >= :crit_runas_events THEN 'CRITICAL'
         WHEN COUNT(*) >= :warn_runas_events THEN 'WARN'
         ELSE 'OK'
       END AS status
FROM system.access.audit
WHERE identity_metadata.run_by IS NOT NULL
  AND identity_metadata.run_as IS NOT NULL
  AND identity_metadata.run_by <> identity_metadata.run_as
  AND event_date >= current_date() - INTERVAL :period_days DAYS
  AND event_date < current_date()
GROUP BY 1, 2, identity_metadata.run_by, identity_metadata.run_as
ORDER BY event_count DESC

SELECT service_name, action_name,
       CASE
         WHEN actor IS NULL OR actor = '__REDACTED__' THEN actor
         WHEN actor LIKE '%@%' THEN concat(substr(actor, 1, 2), '****@****')
         WHEN actor RLIKE '^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$' THEN actor
         ELSE concat(substr(actor, 1, 2), '****')
       END AS actor,
       event_count, distinct_source_ips, first_event_time, last_event_time,
       -- status: worst-first band on admin/role-change event volume per actor+action (field heuristic; :warn_admin_events / :crit_admin_events).
       CASE
         WHEN event_count >= :crit_admin_events THEN 'CRITICAL'
         WHEN event_count >= :warn_admin_events THEN 'WARN'
         ELSE 'OK'
       END AS status
FROM (
  SELECT service_name, action_name,
         COALESCE(user_identity.email, user_identity.subject_name) AS actor,
         COUNT(*) AS event_count,
         COUNT(DISTINCT source_ip_address) AS distinct_source_ips,
         MIN(event_time) AS first_event_time, MAX(event_time) AS last_event_time
  FROM system.access.audit
  WHERE service_name IN ('accounts','accountsAccessControl','unityCatalog')
    AND event_date >= current_date() - INTERVAL :period_days DAYS
    AND event_date < current_date()
  GROUP BY 1, 2, 3
)
ORDER BY event_count DESC

SELECT dc.catalog_name, dc.schema_name, dc.table_name, dc.column_name, dc.class_tag, dc.confidence,
       cm.MASK_NAME,
       CASE WHEN cm.MASK_NAME IS NULL THEN true ELSE false END AS is_unmasked,
       -- status: yes/no risk - a HIGH-confidence classified column with no mask applied.
       CASE WHEN cm.MASK_NAME IS NULL THEN 'CRITICAL' ELSE 'OK' END AS status
FROM system.data_classification.results dc
LEFT JOIN system.information_schema.column_masks cm
  ON  cm.CATALOG_NAME = dc.catalog_name
  AND cm.SCHEMA_NAME  = dc.schema_name
  AND cm.TABLE_NAME   = dc.table_name
  AND cm.COLUMN_NAME  = dc.column_name
WHERE dc.confidence = 'HIGH'
  AND dc.class_tag IS NOT NULL
ORDER BY is_unmasked DESC, dc.catalog_name, dc.schema_name, dc.table_name, dc.column_name

SELECT source_table_full_name, source_column_name, target_table_full_name, target_column_name,
       entity_type, direct_access,
       COUNT(*) AS event_count,
       COUNT(DISTINCT created_by) AS distinct_principals,
       MAX(event_time) AS last_event_time,
       -- status: worst-first band on how many distinct principals drove this column-to-column edge (field heuristic; :warn_reach_principals / :crit_reach_principals).
       CASE
         WHEN COUNT(DISTINCT created_by) >= :crit_reach_principals THEN 'CRITICAL'
         WHEN COUNT(DISTINCT created_by) >= :warn_reach_principals THEN 'WARN'
         ELSE 'OK'
       END AS status
FROM system.access.column_lineage
WHERE source_table_full_name IS NOT NULL
  AND event_date >= current_date() - INTERVAL :period_days DAYS
  AND event_date < current_date()
GROUP BY 1, 2, 3, 4, 5, 6
ORDER BY distinct_principals DESC

-- column_masks lists only manually-applied (ALTER ... SET MASK) masks; ABAC tag-based policy masks never appear here, so an ABAC-governed account can look falsely unmasked - enumerate those via the UC REST API.
SELECT CATALOG_NAME   AS table_catalog,
       SCHEMA_NAME    AS table_schema,
       TABLE_NAME     AS table_name,
       COLUMN_NAME    AS column_name,
       MASK_NAME      AS mask_name,
       MASK_COL_USAGE AS using_columns
FROM system.information_schema.column_masks
ORDER BY table_catalog, table_schema, table_name, column_name

SELECT catalog_name, schema_name, table_name, column_name, class_tag, confidence, data_type,
       MAX(frequency)            AS max_frequency,
       MAX(latest_detected_time) AS latest_detected_time,
       MIN(first_detected_time)  AS first_detected_time
FROM system.data_classification.results
WHERE class_tag IS NOT NULL
GROUP BY 1, 2, 3, 4, 5, 6, 7
ORDER BY catalog_name, schema_name, table_name, column_name

WITH lineage_window AS (
  SELECT *
  FROM system.access.table_lineage
  WHERE event_date >= dateadd(day, -:period_days, current_date())
    AND event_date < current_date()
),
-- Every table that appeared as a lineage SOURCE in the window (any direct_access
-- value - an indirect/view-mediated read of a base table still proves it was read,
-- so we must NOT exclude direct_access=false here or we over-flag base tables).
-- Prefer the discrete *_catalog/_schema/_name columns over CONCAT vs _full_name,
-- because _full_name can be backtick-quoted for reserved-word identifiers.
source_tables AS (
  SELECT DISTINCT
         source_table_catalog AS catalog,
         source_table_schema  AS schema,
         source_table_name    AS name
  FROM lineage_window
  WHERE source_table_name IS NOT NULL
),
-- The managed/external tables this account holds (privilege-scoped inventory).
inventory AS (
  SELECT table_catalog,
         table_schema,
         table_name,
         table_type,
         table_owner,
         created,
         last_altered
  FROM system.information_schema.tables
  WHERE table_catalog <> 'system'
    AND table_schema <> 'information_schema'
    AND table_type IN ('MANAGED', 'EXTERNAL')
)
SELECT inv.table_catalog,
       inv.table_schema,
       inv.table_name,
       inv.table_type,
       CASE
         WHEN inv.table_owner IS NULL OR inv.table_owner = '__REDACTED__' THEN inv.table_owner
         WHEN inv.table_owner LIKE '%@%' THEN concat(substr(inv.table_owner, 1, 2), '****@****')
         WHEN inv.table_owner RLIKE '^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$' THEN inv.table_owner
         ELSE concat(substr(inv.table_owner, 1, 2), '****')
       END AS table_owner,
       inv.created,
       inv.last_altered,
       -- Age in whole days since the object was last altered (NULL -> age unknown).
       datediff(current_date(), DATE(inv.last_altered)) AS days_since_altered,
       -- status: worst-first band on staleness (field heuristic; :warn_dead_days / :crit_dead_days). NULL age -> NOT_ASSESSED, never a finding.
       CASE
         WHEN datediff(current_date(), DATE(inv.last_altered)) IS NULL THEN 'NOT_ASSESSED'
         WHEN datediff(current_date(), DATE(inv.last_altered)) >= :crit_dead_days THEN 'CRITICAL'
         WHEN datediff(current_date(), DATE(inv.last_altered)) >= :warn_dead_days THEN 'WARN'
         ELSE 'OK'
       END AS status
FROM inventory inv
LEFT JOIN source_tables src
  ON  inv.table_catalog = src.catalog
  AND inv.table_schema  = src.schema
  AND inv.table_name    = src.name
WHERE src.name IS NULL          -- never appeared as a lineage source in the window
ORDER BY days_since_altered DESC NULLS LAST

SELECT object_scope, PRIVILEGE_TYPE,
       CASE
         WHEN GRANTEE IS NULL OR GRANTEE = '__REDACTED__' THEN GRANTEE
         WHEN GRANTEE LIKE '%@%' THEN concat(substr(GRANTEE, 1, 2), '****@****')
         WHEN GRANTEE RLIKE '^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$' THEN GRANTEE
         ELSE concat(substr(GRANTEE, 1, 2), '****')
       END AS GRANTEE,
       grant_count, distinct_objects
FROM (
  SELECT 'TABLE' AS object_scope, PRIVILEGE_TYPE, GRANTEE,
         COUNT(*) AS grant_count,
         COUNT(DISTINCT TABLE_CATALOG || '.' || TABLE_SCHEMA || '.' || TABLE_NAME) AS distinct_objects
  FROM system.information_schema.table_privileges
  GROUP BY 1, 2, 3
  UNION ALL
  SELECT 'CATALOG' AS object_scope, PRIVILEGE_TYPE, GRANTEE,
         COUNT(*) AS grant_count,
         COUNT(DISTINCT CATALOG_NAME) AS distinct_objects
  FROM system.information_schema.catalog_privileges
  GROUP BY 1, 2, 3
)
ORDER BY object_scope, PRIVILEGE_TYPE, GRANTEE

-- SAFE as written (uses only GRANTEE/PRIVILEGE_TYPE, shared across all privilege views).
-- NEEDS CONFIRMATION before reading any per-view object-name column: DESCRIBE each sibling view in your own workspace first.
SELECT 'SCHEMA' AS object_scope, PRIVILEGE_TYPE,
       CASE
         WHEN GRANTEE IS NULL OR GRANTEE = '__REDACTED__' THEN GRANTEE
         WHEN GRANTEE LIKE '%@%' THEN concat(substr(GRANTEE, 1, 2), '****@****')
         WHEN GRANTEE RLIKE '^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$' THEN GRANTEE
         ELSE concat(substr(GRANTEE, 1, 2), '****')
       END AS GRANTEE,
       COUNT(*) AS grant_count
FROM system.information_schema.schema_privileges GROUP BY 1,2,GRANTEE
UNION ALL
SELECT 'CONNECTION', PRIVILEGE_TYPE,
       CASE
         WHEN GRANTEE IS NULL OR GRANTEE = '__REDACTED__' THEN GRANTEE
         WHEN GRANTEE LIKE '%@%' THEN concat(substr(GRANTEE, 1, 2), '****@****')
         WHEN GRANTEE RLIKE '^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$' THEN GRANTEE
         ELSE concat(substr(GRANTEE, 1, 2), '****')
       END,
       COUNT(*)
FROM system.information_schema.connection_privileges GROUP BY 1,2,GRANTEE
UNION ALL
SELECT 'CREDENTIAL', PRIVILEGE_TYPE,
       CASE
         WHEN GRANTEE IS NULL OR GRANTEE = '__REDACTED__' THEN GRANTEE
         WHEN GRANTEE LIKE '%@%' THEN concat(substr(GRANTEE, 1, 2), '****@****')
         WHEN GRANTEE RLIKE '^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$' THEN GRANTEE
         ELSE concat(substr(GRANTEE, 1, 2), '****')
       END,
       COUNT(*)
FROM system.information_schema.credential_privileges GROUP BY 1,2,GRANTEE
UNION ALL
SELECT 'EXTERNAL_LOCATION', PRIVILEGE_TYPE,
       CASE
         WHEN GRANTEE IS NULL OR GRANTEE = '__REDACTED__' THEN GRANTEE
         WHEN GRANTEE LIKE '%@%' THEN concat(substr(GRANTEE, 1, 2), '****@****')
         WHEN GRANTEE RLIKE '^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$' THEN GRANTEE
         ELSE concat(substr(GRANTEE, 1, 2), '****')
       END,
       COUNT(*)
FROM system.information_schema.external_location_privileges GROUP BY 1,2,GRANTEE
ORDER BY object_scope, PRIVILEGE_TYPE, GRANTEE

SELECT CASE
         WHEN COALESCE(user_identity.email, user_identity.subject_name) IS NULL OR COALESCE(user_identity.email, user_identity.subject_name) = '__REDACTED__' THEN COALESCE(user_identity.email, user_identity.subject_name)
         WHEN COALESCE(user_identity.email, user_identity.subject_name) LIKE '%@%' THEN concat(substr(COALESCE(user_identity.email, user_identity.subject_name), 1, 2), '****@****')
         WHEN COALESCE(user_identity.email, user_identity.subject_name) RLIKE '^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$' THEN COALESCE(user_identity.email, user_identity.subject_name)
         ELSE concat(substr(COALESCE(user_identity.email, user_identity.subject_name), 1, 2), '****')
       END AS principal,
       source_ip_address, service_name, action_name,
       COUNT(*) AS event_count,
       SUM(CASE WHEN response.status_code = 200 THEN 1 ELSE 0 END) AS success_count,
       SUM(CASE WHEN response.status_code <> 200 OR response.status_code IS NULL THEN 1 ELSE 0 END) AS non_success_count,
       COUNT(DISTINCT source_ip_address) AS distinct_source_ips,
       MIN(event_time) AS first_event_time, MAX(event_time) AS last_event_time,
       -- status: worst-first band on non-success (failed) auth attempts per principal+IP+action (field heuristic; :warn_failed_logins / :crit_failed_logins).
       CASE
         WHEN SUM(CASE WHEN response.status_code <> 200 OR response.status_code IS NULL THEN 1 ELSE 0 END) >= :crit_failed_logins THEN 'CRITICAL'
         WHEN SUM(CASE WHEN response.status_code <> 200 OR response.status_code IS NULL THEN 1 ELSE 0 END) >= :warn_failed_logins THEN 'WARN'
         ELSE 'OK'
       END AS status
FROM system.access.audit
WHERE service_name = 'accounts'
  AND event_date >= current_date() - INTERVAL :period_days DAYS
  AND event_date < current_date()
GROUP BY COALESCE(user_identity.email, user_identity.subject_name), 2, 3, 4
ORDER BY non_success_count DESC

-- NEEDS CONFIRMATION: the source.ip nested subfield name is unverified in this account.
-- A wrong name errors the whole statement - drop or replace it with the verified subfield
-- before running this in your workspace.
SELECT policy_outcome, rule_label, request_path,
       CASE
         WHEN authenticated_as IS NULL OR authenticated_as = '__REDACTED__' THEN authenticated_as
         WHEN authenticated_as LIKE '%@%' THEN concat(substr(authenticated_as, 1, 2), '****@****')
         WHEN authenticated_as RLIKE '^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$' THEN authenticated_as
         ELSE concat(substr(authenticated_as, 1, 2), '****')
       END AS authenticated_as,
       source.ip AS source_ip,
       COUNT(*) AS denial_count,
       MIN(event_time) AS first_event_time, MAX(event_time) AS last_event_time,
       -- status: worst-first band on inbound denial volume per rule+path+principal+IP (field heuristic; :warn_denial_count / :crit_denial_count).
       CASE
         WHEN COUNT(*) >= :crit_denial_count THEN 'CRITICAL'
         WHEN COUNT(*) >= :warn_denial_count THEN 'WARN'
         ELSE 'OK'
       END AS status
FROM system.access.inbound_network
WHERE event_time >= current_timestamp() - INTERVAL :period_days DAYS
GROUP BY 1, 2, 3, authenticated_as, 5
ORDER BY denial_count DESC

SELECT network_source_type, destination_type, access_type, destination,
       dns_event.rcode                AS dns_rcode,
       storage_event.rejection_reason AS storage_rejection_reason,
       COUNT(*) AS denial_count,
       MIN(event_time) AS first_event_time, MAX(event_time) AS last_event_time,
       -- status: worst-first band on outbound denial volume per source+destination combo (field heuristic; :warn_denial_count / :crit_denial_count).
       CASE
         WHEN COUNT(*) >= :crit_denial_count THEN 'CRITICAL'
         WHEN COUNT(*) >= :warn_denial_count THEN 'WARN'
         ELSE 'OK'
       END AS status
FROM system.access.outbound_network
WHERE event_time >= current_timestamp() - INTERVAL :period_days DAYS
GROUP BY 1, 2, 3, 4, 5, 6
ORDER BY denial_count DESC

WITH lineage_window AS (
  SELECT source_table_catalog, source_table_schema, source_table_name, source_column_name,
         target_table_catalog, target_table_schema, target_table_name, target_column_name,
         created_by
  FROM system.access.column_lineage
  WHERE direct_access = true                       -- direct flows only; excludes view expansion
    AND source_column_name IS NOT NULL
    AND target_column_name IS NOT NULL
    AND source_table_catalog <> 'system'
    AND event_date >= dateadd(day, -:period_days, current_date())
    AND event_date < current_date()
),
-- Sensitivity-tagged columns (manual governance tags). We compare on the tag_name
-- against common PII/sensitivity conventions, case-insensitively. Confirm the real
-- tag_name(s) used in your account; if the convention differs this set under-detects,
-- which is the safe direction (we never invent a tag match).
sensitive_tags AS (
  SELECT catalog_name, schema_name, table_name, column_name,
         tag_name, tag_value
  FROM system.information_schema.column_tags
  WHERE lower(tag_name) IN (
          'pii', 'sensitivity', 'sensitive', 'data_classification', 'classification',
          'confidentiality', 'phi', 'pci'
        )
     OR lower(tag_value) IN (
          'pii', 'sensitive', 'confidential', 'restricted', 'phi', 'pci'
        )
),
-- Every column that carries ANY governance tag (used to decide if a TARGET is tagged
-- at all). Join with '=' on the four discrete identifier columns (never LIKE).
any_tagged_column AS (
  SELECT DISTINCT catalog_name, schema_name, table_name, column_name
  FROM system.information_schema.column_tags
)
SELECT lw.source_table_catalog,
       lw.source_table_schema,
       lw.source_table_name,
       lw.source_column_name,
       st.tag_name  AS source_tag_name,
       st.tag_value AS source_tag_value,
       lw.target_table_catalog,
       lw.target_table_schema,
       lw.target_table_name,
       lw.target_column_name,
       CASE
         WHEN lw.created_by IS NULL OR lw.created_by = '__REDACTED__' THEN lw.created_by
         WHEN lw.created_by LIKE '%@%' THEN concat(substr(lw.created_by, 1, 2), '****@****')
         WHEN lw.created_by RLIKE '^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$' THEN lw.created_by
         ELSE concat(substr(lw.created_by, 1, 2), '****')
       END AS created_by,
       COUNT(*) AS event_count,
       -- status: worst-first band on how often this untagged-PII-propagation edge fired (field heuristic; :warn_pii_gap_events / :crit_pii_gap_events).
       CASE
         WHEN COUNT(*) >= :crit_pii_gap_events THEN 'CRITICAL'
         WHEN COUNT(*) >= :warn_pii_gap_events THEN 'WARN'
         ELSE 'OK'
       END AS status
FROM lineage_window lw
-- SOURCE column must be sensitivity-tagged (= join on discrete columns).
JOIN sensitive_tags st
  ON  lw.source_table_catalog = st.catalog_name
  AND lw.source_table_schema  = st.schema_name
  AND lw.source_table_name    = st.table_name
  AND lw.source_column_name   = st.column_name
-- TARGET column must be UNTAGGED: anti-join against any tag (= join, never LIKE).
LEFT JOIN any_tagged_column tt
  ON  lw.target_table_catalog = tt.catalog_name
  AND lw.target_table_schema  = tt.schema_name
  AND lw.target_table_name    = tt.table_name
  AND lw.target_column_name   = tt.column_name
WHERE tt.column_name IS NULL          -- target carries NO governance tag -> propagation gap
GROUP BY lw.source_table_catalog, lw.source_table_schema, lw.source_table_name,
         lw.source_column_name, st.tag_name, st.tag_value,
         lw.target_table_catalog, lw.target_table_schema, lw.target_table_name,
         lw.target_column_name, lw.created_by
ORDER BY event_count DESC

-- It lists only manually-applied (table-level) row filters set via ALTER TABLE; ABAC tag-based policy filters do not appear here, so an ABAC-governed account can look falsely unfiltered - enumerate those via the UC REST API.
SELECT CATALOG_NAME     AS table_catalog,
       SCHEMA_NAME      AS table_schema,
       TABLE_NAME       AS table_name,
       FILTER_NAME      AS filter_name,
       FILTER_COL_USAGE AS target_columns
FROM system.information_schema.row_filters
ORDER BY table_catalog, table_schema, table_name

SELECT source_table_full_name, target_table_full_name, source_type, target_type,
       entity_type, direct_access,
       CASE WHEN source_type IS NOT NULL AND target_type IS NULL THEN 'READ'
            WHEN target_type IS NOT NULL AND source_type IS NULL THEN 'WRITE'
            WHEN source_type IS NOT NULL AND target_type IS NOT NULL THEN 'READ_WRITE'
            ELSE 'UNKNOWN' END AS access_class,
       COUNT(*) AS event_count,
       COUNT(DISTINCT created_by) AS distinct_principals,
       MAX(event_time) AS last_event_time,
       -- status: worst-first band on how many distinct principals drove this source-target edge (field heuristic; :warn_blast_principals / :crit_blast_principals).
       CASE
         WHEN COUNT(DISTINCT created_by) >= :crit_blast_principals THEN 'CRITICAL'
         WHEN COUNT(DISTINCT created_by) >= :warn_blast_principals THEN 'WARN'
         ELSE 'OK'
       END AS status
FROM system.access.table_lineage
WHERE event_date >= current_date() - INTERVAL :period_days DAYS
  AND event_date < current_date()
GROUP BY 1, 2, 3, 4, 5, 6, 7
ORDER BY distinct_principals DESC

SELECT 'COLUMN' AS object_scope, TAG_NAME, TAG_VALUE,
       COUNT(*) AS tagged_object_count
FROM system.information_schema.column_tags
GROUP BY 1, 2, 3
UNION ALL
SELECT 'TABLE' AS object_scope, TAG_NAME, TAG_VALUE,
       COUNT(*) AS tagged_object_count
FROM system.information_schema.table_tags
GROUP BY 1, 2, 3
ORDER BY object_scope, TAG_NAME, TAG_VALUE

SELECT event_date,
       action_name,
       CASE WHEN request_params['endpoint_name'] IS NULL THEN request_params['endpoint_name'] ELSE concat(substr(request_params['endpoint_name'], 1, 2), '****') END AS endpoint_name,
       COUNT(*) AS event_count
FROM system.access.audit
WHERE service_name = 'vectorSearch'
  AND action_name IN (
        'queryVectorIndex', 'queryVectorIndexNextPage',
        'queryVectorIndexRouteOptimized', 'scanVectorIndex',
        'scanVectorIndexRouteOptimized')
  AND event_date >= dateadd(day, -:period_days, current_date())
  AND event_date < current_date()
GROUP BY event_date, action_name, request_params['endpoint_name']
ORDER BY event_date DESC, endpoint_name, action_name

WITH recips AS (
  SELECT SHARE_NAME,
         COUNT(DISTINCT RECIPIENT_NAME) AS recipient_count,
         array_join(collect_set(
           CASE WHEN RECIPIENT_NAME IS NULL OR RECIPIENT_NAME = '__REDACTED__' THEN RECIPIENT_NAME
                WHEN RECIPIENT_NAME LIKE '%@%' THEN concat(substr(RECIPIENT_NAME, 1, 2), '****@****')
                WHEN RECIPIENT_NAME RLIKE '^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$' THEN RECIPIENT_NAME
                ELSE concat(substr(RECIPIENT_NAME, 1, 2), '****') END), ', ') AS recipients
  FROM system.information_schema.share_recipient_privileges
  GROUP BY SHARE_NAME
),
tbls AS (
  SELECT SHARE_NAME, COUNT(*) AS shared_table_count
  FROM system.information_schema.table_share_usage GROUP BY SHARE_NAME
),
schs AS (
  SELECT SHARE_NAME, COUNT(*) AS shared_schema_count
  FROM system.information_schema.schema_share_usage GROUP BY SHARE_NAME
)
SELECT
  s.SHARE_NAME                                              AS share_name,
  CASE WHEN s.SHARE_OWNER IS NULL OR s.SHARE_OWNER = '__REDACTED__' THEN s.SHARE_OWNER
       WHEN s.SHARE_OWNER LIKE '%@%' THEN concat(substr(s.SHARE_OWNER, 1, 2), '****@****')
       WHEN s.SHARE_OWNER RLIKE '^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$' THEN s.SHARE_OWNER
       ELSE concat(substr(s.SHARE_OWNER, 1, 2), '****') END AS share_owner,
  COALESCE(r.recipient_count, 0)                            AS recipient_count,
  r.recipients                                             AS recipients,
  COALESCE(t.shared_table_count, 0)                        AS shared_table_count,
  COALESCE(sc.shared_schema_count, 0)                      AS shared_schema_count,
  s.CREATED                                                AS created,
  CASE
    WHEN COALESCE(r.recipient_count, 0) >= :crit_share_recipients
      OR COALESCE(t.shared_table_count, 0) >= :crit_shared_tables       THEN 'CRITICAL'
    WHEN COALESCE(r.recipient_count, 0) >= :warn_share_recipients       THEN 'WARN'
    ELSE 'OK'
  END AS status
FROM system.information_schema.shares s
LEFT JOIN recips r  ON r.SHARE_NAME  = s.SHARE_NAME
LEFT JOIN tbls   t  ON t.SHARE_NAME  = s.SHARE_NAME
LEFT JOIN schs   sc ON sc.SHARE_NAME = s.SHARE_NAME
ORDER BY
  CASE status WHEN 'CRITICAL' THEN 0 WHEN 'WARN' THEN 1 ELSE 2 END,
  recipient_count DESC, shared_table_count DESC
LIMIT :top_n

SELECT
  t.TABLE_CATALOG AS view_catalog,
  t.TABLE_SCHEMA  AS view_schema,
  t.TABLE_NAME    AS view_name,
  (t.TABLE_TYPE = 'MATERIALIZED_VIEW') AS is_materialized,
  v.IS_UPDATABLE       AS is_updatable,
  v.IS_INSERTABLE_INTO AS is_insertable_into,
  v.SQL_PATH           AS sql_path,
  LENGTH(v.VIEW_DEFINITION) AS definition_chars,
  CASE WHEN upper(v.VIEW_DEFINITION) LIKE '%MASK%' OR upper(v.VIEW_DEFINITION) LIKE '%FILTER%'
       THEN true ELSE false END AS references_mask_or_filter
FROM system.information_schema.tables t
LEFT JOIN system.information_schema.views v
  ON  v.TABLE_CATALOG = t.TABLE_CATALOG
  AND v.TABLE_SCHEMA  = t.TABLE_SCHEMA
  AND v.TABLE_NAME    = t.TABLE_NAME
WHERE t.TABLE_TYPE IN ('VIEW', 'MATERIALIZED_VIEW')
ORDER BY view_catalog, view_schema, view_name
LIMIT :top_n

WITH vt AS (
  SELECT CATALOG_NAME, SCHEMA_NAME, VOLUME_NAME,
         COUNT(*) AS tag_count,
         array_join(collect_set(TAG_NAME), ', ') AS tag_names
  FROM system.information_schema.volume_tags
  GROUP BY CATALOG_NAME, SCHEMA_NAME, VOLUME_NAME
)
SELECT
  v.VOLUME_CATALOG AS volume_catalog,
  v.VOLUME_SCHEMA  AS volume_schema,
  v.VOLUME_NAME    AS volume_name,
  v.VOLUME_TYPE    AS volume_type,
  CASE WHEN v.VOLUME_OWNER IS NULL OR v.VOLUME_OWNER = '__REDACTED__' THEN v.VOLUME_OWNER
       WHEN v.VOLUME_OWNER LIKE '%@%' THEN concat(substr(v.VOLUME_OWNER, 1, 2), '****@****')
       WHEN v.VOLUME_OWNER RLIKE '^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$' THEN v.VOLUME_OWNER
       ELSE concat(substr(v.VOLUME_OWNER, 1, 2), '****') END AS volume_owner,
  v.STORAGE_LOCATION AS storage_location,
  COALESCE(vt.tag_count, 0) AS tag_count,
  vt.tag_names              AS tag_names,
  v.CREATED                 AS created,
  CASE
    WHEN COALESCE(vt.tag_count, 0) = 0 AND upper(v.VOLUME_TYPE) = 'EXTERNAL' THEN 'CRITICAL'
    WHEN COALESCE(vt.tag_count, 0) = 0                                       THEN 'WARN'
    ELSE 'OK'
  END AS status
FROM system.information_schema.volumes v
LEFT JOIN vt
  ON vt.CATALOG_NAME = v.VOLUME_CATALOG AND vt.SCHEMA_NAME = v.VOLUME_SCHEMA AND vt.VOLUME_NAME = v.VOLUME_NAME
ORDER BY
  CASE status WHEN 'CRITICAL' THEN 0 WHEN 'WARN' THEN 1 ELSE 2 END,
  v.VOLUME_CATALOG, v.VOLUME_SCHEMA, v.VOLUME_NAME
LIMIT :top_n

SELECT
  object_type, catalog_name, schema_name, object_name, object_detail, finding, tags,
  CASE
    WHEN is_sensitive = 1 AND is_external = 1 THEN 'CRITICAL'
    WHEN untagged     = 1 AND is_external = 1 THEN 'CRITICAL'
    WHEN is_sensitive = 1                     THEN 'WARN'
    WHEN untagged     = 1                     THEN 'WARN'
    ELSE 'OK'
  END AS status
FROM (
  -- VOLUMES: sensitive-tagged (known PII files) or untagged (blind spot)
  SELECT
    'VOLUME'          AS object_type,
    v.VOLUME_CATALOG  AS catalog_name,
    v.VOLUME_SCHEMA   AS schema_name,
    v.VOLUME_NAME     AS object_name,
    v.VOLUME_TYPE     AS object_detail,
    CASE WHEN upper(v.VOLUME_TYPE) = 'EXTERNAL' THEN 1 ELSE 0 END AS is_external,
    COALESCE(vt.is_sensitive, 0)                                 AS is_sensitive,
    CASE WHEN COALESCE(vt.tag_count, 0) = 0 THEN 1 ELSE 0 END    AS untagged,
    vt.tag_names                                                AS tags,
    CASE WHEN COALESCE(vt.is_sensitive, 0) = 1 THEN 'sensitive-tagged (PII files outside governed tables)'
         WHEN COALESCE(vt.tag_count, 0) = 0    THEN 'untagged (unclassified file store)'
         ELSE 'tagged (non-sensitive)' END                     AS finding
  FROM system.information_schema.volumes v
  LEFT JOIN (
    SELECT CATALOG_NAME, SCHEMA_NAME, VOLUME_NAME,
           COUNT(*) AS tag_count,
           MAX(CASE WHEN TAG_NAME  RLIKE '(?i)(pii|sensitiv|confidential|gdpr|personal|secret|restricted)'
                      OR TAG_VALUE RLIKE '(?i)(pii|sensitiv|confidential|gdpr|personal|secret|restricted)'
                    THEN 1 ELSE 0 END) AS is_sensitive,
           array_join(collect_set(TAG_NAME), ', ') AS tag_names
    FROM system.information_schema.volume_tags
    GROUP BY CATALOG_NAME, SCHEMA_NAME, VOLUME_NAME
  ) vt
    ON vt.CATALOG_NAME = v.VOLUME_CATALOG AND vt.SCHEMA_NAME = v.VOLUME_SCHEMA AND vt.VOLUME_NAME = v.VOLUME_NAME
  UNION ALL
  -- SCHEMAS: only those carrying a sensitivity tag (a namespace flagged sensitive)
  SELECT
    'SCHEMA'          AS object_type,
    st.CATALOG_NAME   AS catalog_name,
    st.SCHEMA_NAME    AS schema_name,
    st.SCHEMA_NAME    AS object_name,
    'schema tag'      AS object_detail,
    0                 AS is_external,
    1                 AS is_sensitive,
    0                 AS untagged,
    array_join(collect_set(concat(st.TAG_NAME, '=', st.TAG_VALUE)), ', ') AS tags,
    'sensitive-tagged schema (namespace flagged sensitive)'              AS finding
  FROM system.information_schema.schema_tags st
  WHERE st.TAG_NAME  RLIKE '(?i)(pii|sensitiv|confidential|gdpr|personal|secret|restricted)'
     OR st.TAG_VALUE RLIKE '(?i)(pii|sensitiv|confidential|gdpr|personal|secret|restricted)'
  GROUP BY st.CATALOG_NAME, st.SCHEMA_NAME
)
WHERE is_sensitive = 1 OR untagged = 1   -- findings only (drop tagged-but-non-sensitive volumes)
ORDER BY
  CASE status WHEN 'CRITICAL' THEN 0 WHEN 'WARN' THEN 1 ELSE 2 END,
  object_type, catalog_name, schema_name, object_name
LIMIT :top_n