Snowflake存储过程报错时如何让调用它的任务标记为失败

问题根因 #

1. 存储过程捕获所有异常后仅返回错误字符串，未抛出未处理异常，Snowflake Task默认将存储过程正常执行返回的场景判定为成功，不会标记失败。
2. 现有代码中system$set_return_value调用存在语法错误，错误信息拼接时格式不正确，导致执行该语句时触发语法报错，覆盖了原始业务错误。
3. 代码存在多处拼写笔误（比如表名smaple_table_temp、字段类型varhcar等），额外引入了编译错误。

修复步骤 #

1. 错误捕获逻辑调整 #

所有错误场景写入日志后，主动抛出异常，而非仅返回错误信息，这样Task才会识别为执行失败。

2. 修正system$set_return_value语法 #

你当前的调用语句中括号、引号未正确闭合，且未正确拼接变量，改用参数绑定的方式传入错误信息，避免字符串拼接导致的语法问题。

3. 修正代码拼写错误 #

• 修正COPY INTO语句中的表名smaple_table_temp为sample_table_temp
• 修正error_table建表语句中的varhcar为varchar
• 修正Task创建语句中的sample task为sample_task（任务名不能包含空格）

完整修正后的存储过程示例 #

create or replace procedure sample_procedure()
returns varchar not null
language javascript
execute as caller    
as
$$
    try
    {
       try
        {
            var ct_table_cmd = `create or replace table sample_table_temp like sample_table`;
            var ct_table_stmt = snowflake.createStatement({sqlText: ct_table_cmd});
            var result_set= ct_table_stmt.execute();
             result_set.next();
        }
        catch(err)
        {
           var queryId = ct_table_stmt.getQueryId();
           var queryText = ct_table_stmt.getSqlText();
           var log_insert_into=snowflake.createStatement({
               sqlText:`insert into error_table
               (code, message, queryid, querytext) VALUES (?,?,?,?);`,
               binds : [err.code, err.message,queryId,queryText]
            });
            log_insert_into.execute();
            // 修正返回值设置语法
            var ct_task_ret_value = snowflake.createStatement({
                sqlText: `call system$set_return_value(?);`,
                binds: [err.message]
            }).execute();
            // 主动抛出异常，触发Task标记为失败
            throw err;
       }
        var copy_cmd = `copy into sample_table_temp from @mystage
                 file_format=(format_name= 'sample_csv_format')
                 files=('/file_name')
                 on_error=skip_file;`;
 

         var copy_cmd_stmt = snowflake.createStatement({sqlText:copy_cmd});
        var result_set =copy_cmd_stmt.execute();
        result_set.next();
       if(result_set.getColumnValue(2)=='LOADED')
       {
          var swap_cmd = `alter table sample_table_temp swap with sample_table;`;
          var swap_stmt = snowflake.createStatement({sqlText: swap_cmd});
          swap_stmt.execute();
          return 'SUCCESS';
       }
       else if (result_set.getColumnValue(2)== 'LOAD_FAILED')
      {
        var err_message= result_set.getColumnValue(7);
        var queryId = copy_cmd_stmt.getQueryId();
        var queryText = copy_cmd_stmt.getSqlText();
        var log_insert_into=snowflake.createStatement({
            sqlText:`insert into error_table
            (code, message, queryid, querytext) VALUES (?,?,?,?);`,
            binds : ['LOAD_FAILED', err_message,queryId,queryText]
          });
        log_insert_into.execute();
        var task_ret_value = snowflake.createStatement({
            sqlText: `call system$set_return_value(?);`,
            binds: [err_message]
        }).execute();
        // 主动抛出自定义错误，触发Task标记为失败
        throw new Error(err_message);
     }
    }
$$
;

验证方法 #

修正后执行如下操作确认效果：

• 直接执行call sample_procedure()触发错误，确认返回的错误信息为预期的业务错误，无语法报错
• 手动触发Task执行，查询SELECT * FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY())，确认错误场景下Task状态为FAILED，且错误信息与直接调用存储过程返回的一致。

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