Snowflake：存储过程执行失败时如何保留流数据

问题根源 #

你当前代码的核心问题是：执行SELECT * FROM STREAM_TEST_1会立即推进流的偏移量——哪怕后续事务回滚，流也不会回溯到之前的状态，导致未成功处理的数据直接从流中消失。

解决方案 #

要确保只有当所有目标表的写入/删除操作全部成功后，才永久推进流的偏移量。可以通过事务包裹全流程+持久化暂存表的方式实现，具体步骤如下：

1. 用持久化暂存表临时存放流数据，避免流被提前消费
2. 将流数据插入、目标表处理、事务提交全流程放在一个事务内
3. 若流程失败则回滚事务，此时插入暂存表的操作会被撤销，流的偏移量不会推进，数据可重新处理

修改后的存储过程代码 #

CREATE OR REPLACE PROCEDURE test_proc()
RETURNS string
LANGUAGE javascript
strict
EXECUTE AS owner
AS
$$
try {
    // 开启命名事务
    snowflake.execute({
        sqlText: "BEGIN NAME test_update;"
    });
    
    // 创建持久化暂存表（结构与流一致），并将流数据插入其中
    snowflake.execute({
        sqlText: "CREATE OR REPLACE TABLE stream_staging LIKE STREAM_TEST_1;"
    });
    snowflake.execute({
        sqlText: "INSERT INTO stream_staging SELECT * FROM STREAM_TEST_1;"
    });
    
    // 从暂存表读取数据，完成各目标表的操作
    snowflake.execute({
        sqlText: "DELETE FROM TAB_2 WHERE ID IN (Select ID from stream_staging where name = 'TEST');"
    });
    
    snowflake.execute({
        sqlText: "DELETE FROM TAB_3 WHERE ID IN (Select ID from stream_staging where name = 'DUMMY');"
    });
    
    snowflake.execute({
        sqlText: "INSERT INTO TAB_4 SELECT ID, NAME FROM stream_staging where name = 'ABC';"
    });

    snowflake.execute({
        sqlText: "INSERT INTO TAB_5 SELECT ID, NAME FROM stream_staging where name = 'XYZ';"
    });     
    
    // 所有操作成功，提交事务——此时流的偏移量会永久推进
    snowflake.execute({
        sqlText: "COMMIT;"
    });
    
    // 可选：清理暂存表
    snowflake.execute({
        sqlText: "TRUNCATE TABLE stream_staging;"
    });
    
    return "success";
} catch (err) {
    // 流程失败，回滚事务——插入暂存表的操作被撤销，流数据不会丢失
    snowflake.execute({
        sqlText: "ROLLBACK;"
    });
    return "failed: " + err.message;
}
$$;

补充说明 #

• 持久化暂存表：用普通表而非临时表，确保事务回滚时数据会被正确撤销，避免会话中断导致数据丢失。
• 事务的核心作用：只有当所有目标表操作都完成且无错误时，才会确认流的消费行为；任何步骤失败都会回滚全流程，流的偏移量保持不变，数据可再次被读取处理。
• 幂等性优化：若需避免重复处理数据，可在暂存表中添加processed_status字段，标记数据是否已完成处理，或基于流的METADATA$ACTION/METADATA$ROW_ID字段做去重判断。

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