如何在BigQuery中从上次失败点重跑存储过程？

BigQuery本身没有原生支持存储过程的断点续跑功能，但可以通过状态追踪表+条件执行逻辑手动实现从失败位置恢复执行的需求。以下是具体的实现方案：

1. 创建状态追踪表 #

首先需要一张表来记录每个子存储过程的执行状态，用于判断哪些步骤已经成功完成，哪些需要重新执行：

CREATE OR REPLACE TABLE `dataset.procedure_execution_status` (
    step_name STRING PRIMARY KEY,
    is_success BOOL,
    executed_at TIMESTAMP
);

2. 初始化状态（可选） #

如果是第一次执行，可以预先插入所有子过程的初始状态（未执行）：

INSERT INTO `dataset.procedure_execution_status` (step_name, is_success, executed_at)
VALUES
    ('procedure1', FALSE, NULL),
    ('procedure2', FALSE, NULL),
    ('procedure3', FALSE, NULL),
    ('procedure4', FALSE, NULL),
    ('procedure5', FALSE, NULL)
ON CONFLICT(step_name) DO NOTHING; -- 避免重复插入

3. 修改主存储过程，加入断点续跑逻辑 #

更新主存储过程，每个子过程调用前先检查状态表，仅当该步骤未成功执行时才调用，同时添加异常捕获来记录失败状态：

CREATE OR REPLACE PROCEDURE `dataset.procedure_name` ()
BEGIN
    -- 处理procedure1
    DECLARE step1_success BOOL;
    SELECT is_success INTO step1_success FROM `dataset.procedure_execution_status` WHERE step_name = 'procedure1';
    IF NOT step1_success THEN
        BEGIN
            CALL `dataset.procedure1`();
            -- 执行成功，更新状态
            UPDATE `dataset.procedure_execution_status` 
            SET is_success = TRUE, executed_at = CURRENT_TIMESTAMP() 
            WHERE step_name = 'procedure1';
        EXCEPTION WHEN ERROR THEN
            -- 执行失败，更新状态记录时间
            UPDATE `dataset.procedure_execution_status` 
            SET executed_at = CURRENT_TIMESTAMP() 
            WHERE step_name = 'procedure1';
            -- 抛出异常终止后续执行，确保不会跳过失败步骤
            RAISE;
        END;
    END IF;

    -- 处理procedure2
    DECLARE step2_success BOOL;
    SELECT is_success INTO step2_success FROM `dataset.procedure_execution_status` WHERE step_name = 'procedure2';
    IF NOT step2_success THEN
        BEGIN
            CALL `dataset.procedure2`();
            UPDATE `dataset.procedure_execution_status` 
            SET is_success = TRUE, executed_at = CURRENT_TIMESTAMP() 
            WHERE step_name = 'procedure2';
        EXCEPTION WHEN ERROR THEN
            UPDATE `dataset.procedure_execution_status` 
            SET executed_at = CURRENT_TIMESTAMP() 
            WHERE step_name = 'procedure2';
            RAISE;
        END;
    END IF;

    -- 处理procedure3
    DECLARE step3_success BOOL;
    SELECT is_success INTO step3_success FROM `dataset.procedure_execution_status` WHERE step_name = 'procedure3';
    IF NOT step3_success THEN
        BEGIN
            CALL `dataset.procedure3`();
            UPDATE `dataset.procedure_execution_status` 
            SET is_success = TRUE, executed_at = CURRENT_TIMESTAMP() 
            WHERE step_name = 'procedure3';
        EXCEPTION WHEN ERROR THEN
            UPDATE `dataset.procedure_execution_status` 
            SET executed_at = CURRENT_TIMESTAMP() 
            WHERE step_name = 'procedure3';
            RAISE;
        END;
    END IF;

    -- 处理procedure4
    DECLARE step4_success BOOL;
    SELECT is_success INTO step4_success FROM `dataset.procedure_execution_status` WHERE step_name = 'procedure4';
    IF NOT step4_success THEN
        BEGIN
            CALL `dataset.procedure4`();
            UPDATE `dataset.procedure_execution_status` 
            SET is_success = TRUE, executed_at = CURRENT_TIMESTAMP() 
            WHERE step_name = 'procedure4';
        EXCEPTION WHEN ERROR THEN
            UPDATE `dataset.procedure_execution_status` 
            SET executed_at = CURRENT_TIMESTAMP() 
            WHERE step_name = 'procedure4';
            RAISE;
        END;
    END IF;

    -- 处理procedure5
    DECLARE step5_success BOOL;
    SELECT is_success INTO step5_success FROM `dataset.procedure_execution_status` WHERE step_name = 'procedure5';
    IF NOT step5_success THEN
        BEGIN
            CALL `dataset.procedure5`();
            UPDATE `dataset.procedure_execution_status` 
            SET is_success = TRUE, executed_at = CURRENT_TIMESTAMP() 
            WHERE step_name = 'procedure5';
        EXCEPTION WHEN ERROR THEN
            UPDATE `dataset.procedure_execution_status` 
            SET executed_at = CURRENT_TIMESTAMP() 
            WHERE step_name = 'procedure5';
            RAISE;
        END;
    END IF;
END

4. 关键说明 #

• 状态重置：如果需要重新执行某一步骤，只需将该步骤的is_success字段更新为FALSE即可。
• 异常处理：示例中在子过程失败时抛出异常终止后续执行，避免跳过关键步骤导致数据不一致；若有特殊需求，可调整逻辑，但需谨慎评估风险。
• 幂等性：确保每个子存储过程是幂等的（重复执行不会产生错误或重复数据），否则断点续跑可能导致数据异常。

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