如何在Snowflake存储过程中执行LIST命令或等待任务完成？

1. 解决存储过程中LIST命令报错的问题 #

Snowflake SQL存储过程不直接支持LIST这类元数据命令，但可以通过**动态SQL+RESULT_SCAN**的方式间接获取文件列表：

CREATE OR REPLACE PROCEDURE enumerate_dms_files(p_stage_name VARCHAR)
RETURNS TABLE (name VARCHAR, size NUMBER, md5 VARCHAR, last_modified TIMESTAMP_LTZ)
LANGUAGE SQL
AS
$$
DECLARE
    list_query_id VARCHAR;
BEGIN
    -- 执行LIST命令并捕获查询ID
    EXECUTE IMMEDIATE 'LIST @' || p_stage_name INTO list_query_id;
    -- 通过RESULT_SCAN提取LIST的结果集
    RETURN TABLE(RESULT_SCAN(list_query_id));
END;
$$;

说明 #

• 用EXECUTE IMMEDIATE执行LIST命令，将查询ID存入变量
• 通过RESULT_SCAN()函数获取该查询的结果，转换为表返回
• 你可以在存储过程中遍历这个结果集，添加自定义逻辑（比如筛选特定文件、记录日志等）

2. 存储过程等待任务完成（无需任务链） #

通过循环查询INFORMATION_SCHEMA.TASK_HISTORY，监控任务的执行状态，直到任务进入终态（成功/失败/取消）或超时：

CREATE OR REPLACE PROCEDURE wait_for_task_completion(p_task_name VARCHAR, p_timeout_minutes NUMBER DEFAULT 30)
RETURNS VARCHAR
LANGUAGE SQL
AS
$$
DECLARE
    v_task_state VARCHAR;
    v_start_time TIMESTAMP_LTZ := CURRENT_TIMESTAMP();
BEGIN
    LOOP
        -- 获取任务最新一次执行的状态
        SELECT state INTO v_task_state
        FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(
            task_name => p_task_name,
            result_limit => 1
        ))
        WHERE scheduled_time = (SELECT MAX(scheduled_time) FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(task_name => p_task_name)));

        -- 检查任务是否完成
        IF v_task_state IN ('SUCCEEDED', 'FAILED', 'CANCELLED') THEN
            RETURN '任务 [' || p_task_name || '] 执行完成，状态: ' || v_task_state;
        END IF;

        -- 检查是否超时
        IF CURRENT_TIMESTAMP() > DATEADD('minute', p_timeout_minutes, v_start_time) THEN
            RETURN '等待任务 [' || p_task_name || '] 超时';
        END IF;

        -- 等待5秒后再轮询（可根据实际调整间隔）
        CALL SYSTEM$WAIT(5);
    END LOOP;
END;
$$;

说明 #

• 每次轮询获取任务最新的执行状态，避免重复查询历史记录
• 使用SYSTEM$WAIT()降低轮询频率，减少资源消耗
• 需要确保存储过程拥有访问INFORMATION_SCHEMA.TASK_HISTORY的权限

3. 整合场景示例 #

在同一个存储过程中完成文件枚举、触发任务、等待任务完成的完整流程：

CREATE OR REPLACE PROCEDURE orchestrate_dms_load(p_stage_name VARCHAR, p_task_name VARCHAR)
RETURNS VARCHAR
LANGUAGE SQL
AS
$$
DECLARE
    v_file_list RESULTSET;
BEGIN
    -- 1. 枚举DMS文件
    v_file_list := (SELECT * FROM TABLE(enumerate_dms_files(p_stage_name)));
    
    -- 可选：记录文件列表到日志表
    INSERT INTO dms_file_load_log(file_name, last_modified, load_timestamp)
    SELECT name, last_modified, CURRENT_TIMESTAMP() FROM TABLE(v_file_list);

    -- 2. 手动触发任务（如果任务是暂停状态，先恢复）
    EXECUTE IMMEDIATE 'ALTER TASK ' || p_task_name || ' RESUME';
    EXECUTE IMMEDIATE 'EXECUTE TASK ' || p_task_name;

    -- 3. 等待任务完成
    RETURN CALL wait_for_task_completion(p_task_name);
END;
$$;

注意事项 #

• 存储过程需要拥有EXECUTE TASK、ALTER TASK以及访问目标阶段的权限
• 如果任务是按固定调度运行的，需确保手动触发不会与自动调度冲突
• 可根据实际业务调整超时时间和轮询间隔

内容的提问来源于stack exchange，提问作者Simeon Pilgrim