Airflow 2.5.1调用Snowflake JS存储过程遇事务回滚错误求助

错误原因 #

1. Snowflake事务机制冲突：Snowflake存储过程默认处于自动提交模式，显式执行BEGIN TRANSACTION会创建用户级范围事务。若存储过程结束时该事务未被显式提交/回滚，Snowflake会自动回滚并抛出此错误。
2. 异常处理逻辑漏洞：当BEGIN TRANSACTION执行失败时，后续的ROLLBACK会尝试回滚不存在的事务，触发额外异常，导致事务状态无法正确收尾。
3. Airflow自动提交冲突：SnowflakeOperator默认autocommit=True，会在SQL执行完成后自动提交事务，与存储过程内部的手动事务管理产生冲突，加剧事务状态混乱。

解决方法 #

方法1：移除存储过程内的手动事务管理（推荐） #

Snowflake中单个DML/DDL语句默认自动提交，除非需要多语句原子性，否则无需手动管控事务。修改后的存储过程代码：

try {
    // 写入你的核心业务SQL逻辑
    // snowflake.execute({sqlText: "INSERT/UPDATE/DELETE ..."});
}
catch(err) {
    // 直接抛出异常，让调用方感知错误
    throw err;
}

方法2：保留手动事务并完善逻辑+调整Airflow配置 #

若必须手动管理事务，需确保事务完全闭环，同时处理无活跃事务时的回滚场景，并关闭Airflow的自动提交：

步骤1：修改存储过程代码

try {
    snowflake.execute({sqlText: "BEGIN TRANSACTION;"});
    // 核心业务SQL操作
    // snowflake.execute({sqlText: "你的业务SQL"});
    snowflake.execute({sqlText: "COMMIT;"});
}
catch(err) {
    // 先检查是否存在活跃事务，再执行回滚
    let rs = snowflake.execute({sqlText: "SELECT CURRENT_TRANSACTION()"});
    if (rs.next() && rs.getColumnValue(1) != null) {
        snowflake.execute({sqlText: "ROLLBACK;"});
    }
    throw err;
}

步骤2：调整Airflow的SnowflakeOperator配置

添加autocommit=False参数，让存储过程内部自主管理事务：

call_CONFIG_DATA_LOAD = SnowflakeOperator(
    task_id='call_CONFIG_DATA_LOAD',
    snowflake_conn_id=snowflake_conn_id,
    database=database_name,
    schema=target_schema,
    role=admin_role,
    warehouse=warehouse,
    sql=create_config_table,
    autocommit=False,
    dag=dag
)

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