如何监控DDL语句？创建INFORMATION_SCHEMA.COLUMNS Stream遭拒求解决方案

方法1：利用事件表（Event Tables）捕获全量DDL操作 #

Snowflake的事件表会记录账户或数据库级别的所有操作事件，包括各类DDL语句，是最直接的监控方式：

• 先启用事件表（账户级或指定数据库）：

  -- 账户级启用
  ALTER ACCOUNT SET EVENT_TABLE = YOUR_DB.YOUR_SCHEMA.EVENT_LOG_TABLE;
  -- 或仅针对特定数据库启用
  ALTER DATABASE YOUR_DB SET EVENT_TABLE = YOUR_DB.YOUR_SCHEMA.EVENT_LOG_TABLE;
  

• 查询事件表过滤DDL操作：

  SELECT EVENT_TIMESTAMP, EVENT_TYPE, QUERY_TEXT, USER_NAME, OBJECT_NAME
  FROM YOUR_DB.YOUR_SCHEMA.EVENT_LOG_TABLE
  WHERE EVENT_TYPE IN ('CREATE_TABLE', 'ALTER_TABLE', 'DROP_TABLE', 'CREATE_SCHEMA', 'ALTER_SCHEMA', 'DROP_SCHEMA') -- 按需添加DDL类型
  ORDER BY EVENT_TIMESTAMP DESC;
  

• 结合Task定期处理：创建按时间间隔执行的Task，扫描事件表中的新增DDL记录，触发对应存储过程或后续逻辑。

方法2：自定义元数据表+Stream监控元数据变化 #

由于无法直接在内置视图上创建Stream，可通过同步元数据到自定义表，再基于该表创建Stream：

1. 创建存储元数据的自定义表：

   CREATE TABLE CUSTOM_COLUMN_METADATA (
       TABLE_CATALOG VARCHAR,
       TABLE_SCHEMA VARCHAR,
       TABLE_NAME VARCHAR,
       COLUMN_NAME VARCHAR,
       DATA_TYPE VARCHAR,
       LAST_ALTERED TIMESTAMP,
       LOAD_TIMESTAMP TIMESTAMP DEFAULT CURRENT_TIMESTAMP()
   );
   

2. 创建Task定期同步INFORMATION_SCHEMA.COLUMNS数据：

   CREATE OR REPLACE TASK SYNC_COLUMN_META_TASK
   WAREHOUSE = YOUR_WH
   SCHEDULE = 'USING CRON 5 * * * * UTC' -- 每小时第5分钟执行一次
   AS
   MERGE INTO CUSTOM_COLUMN_METADATA t
   USING (
       SELECT TABLE_CATALOG, TABLE_SCHEMA, TABLE_NAME, COLUMN_NAME, DATA_TYPE, LAST_ALTERED
       FROM INFORMATION_SCHEMA.COLUMNS
   ) s
   ON t.TABLE_CATALOG = s.TABLE_CATALOG 
   AND t.TABLE_SCHEMA = s.TABLE_SCHEMA 
   AND t.TABLE_NAME = s.TABLE_NAME 
   AND t.COLUMN_NAME = s.COLUMN_NAME
   WHEN MATCHED AND t.LAST_ALTERED <> s.LAST_ALTERED THEN 
       UPDATE SET t.DATA_TYPE = s.DATA_TYPE, t.LAST_ALTERED = s.LAST_ALTERED, t.LOAD_TIMESTAMP = CURRENT_TIMESTAMP()
   WHEN NOT MATCHED THEN 
       INSERT (TABLE_CATALOG, TABLE_SCHEMA, TABLE_NAME, COLUMN_NAME, DATA_TYPE, LAST_ALTERED)
       VALUES (s.TABLE_CATALOG, s.TABLE_SCHEMA, s.TABLE_NAME, s.COLUMN_NAME, s.DATA_TYPE, s.LAST_ALTERED);
   

3. 在自定义表上创建Stream捕获变化：

   CREATE OR REPLACE STREAM COLUMN_META_STREAM
   ON TABLE CUSTOM_COLUMN_METADATA
   APPEND_ONLY = FALSE; -- 支持捕获增、删、改变化
   

4. 创建Task处理Stream中的变化：

   CREATE OR REPLACE TASK PROCESS_META_CHANGES_TASK
   WAREHOUSE = YOUR_WH
   AFTER SYNC_COLUMN_META_TASK
   AS
   CALL YOUR_DDL_HANDLER_PROCEDURE();
   

方法3：通过告警功能触发DDL相关操作 #

利用Snowflake的告警功能，基于事件表的查询设置规则，当检测到目标DDL操作时自动触发存储过程或通知：

• 示例告警规则（检测DROP TABLE操作）：

  CREATE ALERT DROP_TABLE_ALERT
  WAREHOUSE = YOUR_WH
  SCHEDULE = 'USING CRON * * * * * UTC' -- 每分钟检查一次
  WHEN (
      SELECT COUNT(*)
      FROM YOUR_DB.YOUR_SCHEMA.EVENT_LOG_TABLE
      WHERE EVENT_TYPE = 'DROP_TABLE'
      AND EVENT_TIMESTAMP >= CURRENT_TIMESTAMP() - INTERVAL '1 MINUTE'
  ) > 0
  THEN CALL BACKUP_DROPPED_TABLE_PROCEDURE();
  

内容的提问来源于stack exchange，提问作者Eric Mamet