UDF 错误处理与重试

场景描述

在AI数据工程中，UDF常涉及网络调用、外部模型推理、数据解析等易出错操作。当UDF执行遇到异常时，默认行为是直接抛出异常并中断整个查询。但在实际业务中，用户往往需要更灵活的错误处理策略，例如：

• 静默忽略：部分数据异常不影响整体结果，希望跳过错误行继续执行。
• 日志记录：记录错误信息用于事后排查，但不中断查询。
• 自动重试：网络超时、服务临时不可用等场景，希望自动重试以提升成功率。
• 按异常类型区分：不同类型的异常采用不同策略，例如超时重试、数据格式错误忽略。

UDF 错误处理与重试功能允许用户在不修改源UDF代码的前提下，通过 wrap_function API 为已注册的UDF动态包装错误处理能力，支持忽略、日志记录、指定类型重试、退避策略等组合配置。

约束限制

• 仅支持内置异常类型：retry_on、ignore_on、log_on中指定的异常类型必须为 Python 内置异常（如 ValueError、TimeoutError、RuntimeError 等），不支持自定义异常类。
• 不支持包装BUILTIN函数：wrap_function 仅支持包装Python UDF，不支持数据库内置函数（如 ABS()、UPPER() 等）。
• WRAPPER默认为TEMPORARY：包装后的UDF默认为临时函数，会话结束后自动销毁。
• Vectorized UDF整批失败：对于PyarrowVector / PandasVector 类型的UDF，当批次中任一行触发异常时，整个批次返回None。

配置项说明

通过 ErrorHandlerConfig 配置错误处理行为，各参数说明如下：

错误处理优先级

当UDF执行遇到异常时，按以下优先级依次匹配：

┌─ 优先级 1: ignore_on（最高）
│   if ignore_on and isinstance(e, ignore_on):
│       return None                    ← 静默忽略
│
├─ 优先级 2: log_on
│   if log_on and isinstance(e, log_on):
│       _log_error(e, ...)
│       return None                    ← 记录日志 + 返回 None
│
├─ 优先级 3: retry_on（指定类型重试）
│   if retry_on and isinstance(e, retry_on)
│           and retry_count < max_retries:
│       delay = _compute_delay(...)
│       sleep(delay)
│       continue                       ← 重试
│
├─ 优先级 4: on_error=RETRY（兜底全重试）
│   if on_error==RETRY and not retry_on
│           and retry_count < max_retries:
│       delay = _compute_delay(...)
│       sleep(delay)
│       continue                       ← 重试所有异常
│
├─ 优先级 5: on_error=IGNORE
│   if on_error == IGNORE:
│       return None                    ← 静默返回 None
│
├─ 优先级 6: on_error=LOG
│   if on_error == LOG:
│       _log_error(e, ...)
│       return None                    ← 记录日志 + 返回 None
│
└─ 优先级 7: raise（默认 / 重试耗尽）
    raise                              ← 抛出原始异常

关键规则

• ignore_on优先级最高，匹配后直接返回None，不进入重试或日志逻辑。
• log_on优先级高于retry_on，匹配后记录日志并返回None，不进入重试。
• retry_on指定类型重试；on_error=RETRY不指定类型，对所有异常重试。两者共存时retry_on优先。
• 重试耗尽后，落入on_error决定最终行为（RAISE/IGNORE/LOG）。

退避策略

重试间隔由backoff策略和initial_delay_ms计算，所有策略最终受max_delay_ms约束：

所有策略最终受max_delay_ms约束：min(delay, max_delay_ms)。

使用方式

1. 注册源UDF：按照正常流程注册Python UDF（Scalar / Class / PyarrowVector / PandasVector 均可）。
2. 调用wrap_function包装：通过AuraConnection.wrap_function()为源 UDF 添加错误处理配置，返回一个包装后的可调用对象。

   wrapped = connwrap_function(
       source_udf_name="my_func",       # 源 UDF 名称
       source_database="my_db",          # 源 UDF 所在数据库
       config=ErrorHandlerConfig(...),   # 错误处理配置
       wrapper_name="my_func_wrapped",   # 包装后 UDF 名称（可选）
    )

3. 使用包装后的UDF执行查询：将包装后的UDF用于ds.map() 或 df.select()等操作。

   ds = connload_dataset("my_table", database="my_db")
   result_ds = ds.map(wrapped, on="input_col", as_col="result")
   df = result_ds.execute()

示例

示例1：指定异常类型重试

网络调用UDF，遇到TimeoutError时自动指数退避重试，重试最多3次：

from aura_frame import AuraConnection, ErrorHandlerConfig, OnErrorAction, BackoffStrategy
con = AuraConnection(...)
wrapped = connwrap_function(
    source_udf_name="call_api",
    source_database="my_db",
    config=ErrorHandlerConfig(
        retry_on=[TimeoutError],
        max_retries=3,
        backoff=BackoffStrategy.EXPONENTIAL,
        initial_delay_ms=200,
        max_delay_ms=5000,
    ),
)
ds = connload_dataset("api_input", database="my_db")
result_ds = ds.map(wrapped, on="input", as_col="result")
df = result_ds.execute()

示例2：多策略组合

源UDF同时支持重试、日志降级和静默忽略，异常匹配按 ignore_on 、log_on 、retry_on 、on_error(默认RAISE) 的优先级依次判断：

1. IndexError：静默忽略返回 None，不记录日志、不重试。
2. ValueError/TypeError：记录日志后返回 None，因配置了 local_log_path="/tmp/udf_errors.log"，写日志到该目录下。
3. ConnectionError/TimeoutError：
   1. 重试最多5次，每次重试前：记录日志（同上）→ 计算延迟 → sleep → 重试。
   2. 当5次重试全部失败后再次异常：on_error默认为RAISE，抛出原始异常，该行执行失败。
4. 其他未配置的异常（如 RuntimeError、KeyError 等）：on_error 默认为 RAISE，抛出原始异常，该行执行失败。

wrapped = connwrap_function(
    source_udf_name="complex_model",
    source_database="my_db",
    config=ErrorHandlerConfig(
        retry_on=[ConnectionError, TimeoutError],
        log_on=[ValueError, TypeError],
        ignore_on=[IndexError],
        max_retries=5,
        backoff=BackoffStrategy.LINEAR,
        initial_delay_ms=500,
        local_log_path="/tmp/udf_errors.log",
    ),
)