# UDF 错误处理与重试
#### 场景描述
在AI数据工程中，UDF常涉及网络调用、外部模型推理、数据解析等易出错操作。当UDF执行遇到异常时，默认行为是直接抛出异常并中断整个查询。但在实际业务中，用户往往需要更灵活的错误处理策略，例如：
- 静默忽略：部分数据异常不影响整体结果，希望跳过错误行继续执行。
- 日志记录：记录错误信息用于事后排查，但不中断查询。
- 自动重试：网络超时、服务临时不可用等场景，希望自动重试以提升成功率。
- 按异常类型区分：不同类型的异常采用不同策略，例如超时重试、数据格式错误忽略。
UDF 错误处理与重试功能允许用户在不修改源UDF代码的前提下，通过 wrap_function API 为已注册的UDF动态包装错误处理能力，支持忽略、日志记录、指定类型重试、退避策略等组合配置。
#### 约束限制
- **仅支持内置异常类型** ：retry_on、ignore_on、log_on中指定的异常类型必须为 [Python 内置异常](https://docs.python.org/3.11/library/exceptions.html)（如 ValueError、TimeoutError、RuntimeError 等），不支持自定义异常类。
- **不支持包装BUILTIN函数**：wrap_function 仅支持包装Python UDF，不支持数据库内置函数（如 ABS()、UPPER() 等）。
- **WRAPPER默认为TEMPORARY**：包装后的UDF默认为临时函数，会话结束后自动销毁。
- **Vectorized UDF整批失败**：对于PyarrowVector / PandasVector 类型的UDF，当批次中任一行触发异常时，整个批次返回None。
 
#### 配置项说明
通过 ErrorHandlerConfig 配置错误处理行为，各参数说明如下：
表1ErrorHandlerConfig参数说明 
| 参数名              | 类型                        | 说明                                                                                                                                                                                                                                                                                                                                                                                                                                                                                             | 默认值   |
|:---|:---|:---|:---|
| on_error         | OnErrorAction             | 全局错误策略。 - RAISE：抛出异常（默认）。  - IGNORE：静默返回None。  - LOG：记录日志并返回None。  - RETRY：对所有异常重试。   | RAISE |
| retry_on         | list\[type\[Exception\]\] | 触发重试的异常类型白名单。匹配时进入重试逻辑，不匹配时落入 on_error 兜底。                                                                                                                                                                                                                                                                                                                                                                                                                                                     | None  |
| ignore_on        | list\[type\[Exception\]\] | 触发静默忽略的异常类型白名单。匹配时直接返回None，**优先级最高**。                                                                                                                                                                                                                                                                                                                                                                                                                                                          | None  |
| log_on           | list\[type\[Exception\]\] | 触发日志记录的异常类型白名单。匹配时记录日志并返回None。                                                                                                                                                                                                                                                                                                                                                                                                                                                                 | None  |
| max_retries      | int                       | 最大重试次数。                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        | 0     |
| backoff          | BackoffStrategy           | 退避策略。 - FIXED：固定间隔。  - LINEAR：线性递增。  - EXPONENTIAL：指数递增。                                                                                                                      | FIXED |
| initial_delay_ms | int                       | 初始延迟，单位为毫秒，作为退避计算的基数。                                                                                                                                                                                                                                                                                                                                                                                                                                                                          | 100   |
| max_delay_ms     | int                       | 延迟上限，单位为毫秒。所有退避策略计算结果受此约束。                                                                                                                                                                                                                                                                                                                                                                                                                                                                     | 60000 |
| local_log_path   | str                       | 本地错误日志文件路径。为None时走Python logger输出warning。                                                                                                                                                                                                                                                                                                                                                                                                                                                      | None  |
| obs_log_config   | ObsLogConfig              | OBS日志上传配置，包含obs_server、obs_bucket_name、obs_directory_base。                                                                                                                                                                                                                                                                                                                                                                                                                                     | None  |
   
#### 错误处理优先级
当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约束：
表2退避策略间隔时间说明 
| backoff 策略  | 计算公式                                     | 示例（initial=100ms, max=5000ms）       |
|:---|:---|:---|
| FIXED       | delay = initial_delay_ms                 | 100, 100, 100, ...                  |
| LINEAR      | delay = initial_delay_ms × (attempt + 1) | 100, 200, 300, 400, ...             |
| EXPONENTIAL | delay = initial_delay_ms × 2\^attempt    | 100, 200, 400, 800, 1600, 3200, ... |
   
所有策略最终受max_delay_ms约束：min(delay, max_delay_ms)。
#### 使用方式
1. **注册源UDF**：按照正常流程注册Python UDF（Scalar / Class / PyarrowVector / PandasVector 均可）。
2. **调用wrap_function包装** ：通过AuraConnection.wrap_function()为源 UDF 添加错误处理配置，返回一个包装后的可调用对象。
   ```
   wrapped = con.wrap_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 = con.load_dataset("my_table", database="my_db")
   result_ds = ds.map(wrapped, on="input_col", as_col="result")
   df = result_ds.execute()
   con.close()
   ```
   
 
#### 示例
**示例1：指定异常类型重试**
网络调用UDF，遇到TimeoutError时自动指数退避重试，重试最多3次：
```
from aura_frame import ErrorHandlerConfig, BackoffStrategy
from aura_frame.multimodal import ai_lake
con = ai_lake.connect(
    aura_endpoint="your-endpoint:port",
    aura_workspace_id="your-workspace-id",
    lf_catalog_name="your_catalog",
    access_key="your-ak",
    secret_key="your-sk",
    default_database="my_db",
    use_single_cn_mode=True,
)
try:
    wrapped = con.wrap_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 = con.load_dataset("api_input", database="my_db")
    result_ds = ds.map(wrapped, on="input", as_col="result")
    df = result_ds.execute()
finally:
    con.close()
```
**示例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 = con.wrap_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",
    ),
)
```