User-Defined Function APIs

UDF显式注册语法

显式注册的含义是用户需要手动在Python代码中侵入式添加注册逻辑代码，需要用户使用backend...register/register_from_file来实现，调用即注册。显式注册依赖于已经获得backend会话对象才能进行。

推荐使用显式注册的场景：如果用户希望明确控制注册时间，允许侵入式添加注册逻辑，或对同一个Backend连接下的UDF注册和使用分离有要求。

一个典型的场景是1个开发团队负责UDF的注册，多个团队负责UDF的使用，注册团队和使用团队之间的Python脚本不互通。

UDF隐式注册语法

隐式注册的含义是依赖Python运行时自动发现并注册UDF。用户不需要在Python代码中侵入式添加注册逻辑代码，而是使用@装饰器修饰原始Python函数，然后在DataFrame中使用被装饰的原始Python函数的标识符，即可完成注册。隐式注册在使用@装饰器时不需要获得backend会话对象，后续在ibis DataFrame中获取backend会话对象。

推荐使用隐式注册的场景：如果用户希望无侵入式地注册UDF，且对同一个Backend连接下的UDF注册和使用分离无要求。

一个典型的场景是1个用户的Python脚本中直接写了整个UDF的注册和使用全流程。

对于隐式注册，实际注册动作的发生时间根据DataFrame操作模式Lazy、Eager有所区别。

参考上文提及的ibis官方文档，对于DataFrame的操作分为Eager、Lazy两种模式，由配置项ibis.options.interactive控制，默认为False，即所有的DataFrame都默认是Lazy模式。对于两种DataFrame执行模式，UDF注册的发生时间不同，详情说明如下：

Python/Pyarrow/Pandas UDF注册参数

注册Python/Pyarrow/Pandas UDF的作用是将一个原始的Python函数或类注册进数据库中。

不管是显式注册还是隐式注册，不管是Scalar UDF、UDAF、UDTF，对于注册Python、Pyarrow、Pandas类型的UDF，目前都接受用户传入以下参数：

Python/Pyarrow/Pandas UDF注册参数注意事项

• 对于import参数，只支持用户传入当前Python函数或类所在的.py文件同级目录或子目录下的文件路径。
• 对于fn参数，如果fn不在当前注册UDF的.py文件中，那么需要同时在imports参数中添加fn定义的文件路径，例如：

  from process import outer
  from aura_frame.multimodal import ai_lake
  
  con = ai_lake.connect(
          aura_endpoint=os.getenv("aura_endpoint"),
          aura_endpoint_name=os.getenv("aura_endpoint_name"),
          aura_workspace_id=os.getenv("aura_workspace_id"),
          lf_catalog_name=os.getenv("lf_catalog_name"),
          access_key=os.getenv("access_key"),
          secret_key=os.getenv("secret_key"),
          default_database=target_database,
          use_single_cn_mode=True
  )
  
  # 注册UDAF
  udf = connudaf.python.register(
      outer(), #从外部引入的fn
      imports=["process.py"] #为fn添加文件路径
  )

• 对于signature参数，目前允许用户传入，用户传入的优先级高于自动推断。当用户不传入时，进入参数/返回值类型自动推断逻辑，详情请参见signature参数的类型推断。
  • 对于注册Pyarrow UDF，无论用户是否传入signature参数，都需要依赖PyarrowVector，例如：

    import aura_frame as aura
    import pyarrow as pa
    import pyarrow.compute as pc
    from aura_frame.multimodal import ai_lake
    
    # === 用户传入signature，需要依赖PyarrowVector ===
    def calculate_sum(
        prices: pa.ChunkedArray,
        quantities: pa.ChunkedArray,
    ) -> pa.ChunkedArray:
        return pc.multiply(prices, quantities)
    
    con = ai_lake.connect(
            aura_endpoint=os.getenv("aura_endpoint"),
            aura_endpoint_name=os.getenv("aura_endpoint_name"),
            aura_workspace_id=os.getenv("aura_workspace_id"),
            lf_catalog_name=os.getenv("lf_catalog_name"),
            access_key=os.getenv("access_key"),
            secret_key=os.getenv("secret_key"),
            default_database=target_database,
            use_single_cn_mode=True
    )
    
    # 注册UDF
    conncreate_scalar_function(
            calculate_sum,
            database=target_database
        )
    
    # === 用户不传入signature，也需要依赖PyarrowVector ===
    def calculate_sum(
        prices: aura.PyarrowVector[float],
        quantities: aura.PyarrowVector[int],
    ) -> aura.PyarrowVector[float]:
        return aura.PyarrowVector[float](pc.multiply(prices, quantities))
    
    con = ai_lake.connect(
            aura_endpoint=os.getenv("aura_endpoint"),
            aura_endpoint_name=os.getenv("aura_endpoint_name"),
            aura_workspace_id=os.getenv("aura_workspace_id"),
            lf_catalog_name=os.getenv("lf_catalog_name"),
            access_key=os.getenv("access_key"),
            secret_key=os.getenv("secret_key"),
            default_database=target_database,
            use_single_cn_mode=True
    )
    
    # 注册UDF
    conncreate_scalar_function(
            calculate_sum,
            database=target_database
        )

  • 对于注册Pandas UDF，无论用户是否传入signature参数，都需要依赖PandasVector，例如：

    import aura_frame as aura
    import pandas as pd
    from aura_frame.multimodal import ai_lake
    
    # === 用户传入signature，需要依赖PandasVector ===
    def calculate_sum(
        prices: pd.Series, 
        quantities: pd.Series,
    ) -> pd.Series:
        return pd.Series(prices * quantities, dtype=pd.Float64Dtype())
    
    con = ai_lake.connect(
            aura_endpoint=os.getenv("aura_endpoint"),
            aura_endpoint_name=os.getenv("aura_endpoint_name"),
            aura_workspace_id=os.getenv("aura_workspace_id"),
            lf_catalog_name=os.getenv("lf_catalog_name"),
            access_key=os.getenv("access_key"),
            secret_key=os.getenv("secret_key"),
            default_database=target_database,
            use_single_cn_mode=True
    )
    
    # 注册UDF
    conncreate_scalar_function(
            calculate_sum,
            database=target_database
        )
    
    # === 用户不传入signature，也需要依赖PandasVector ===
    def calculate_sum(
        prices: aura.PandasVector[float], 
        quantities: aura.PandasVector[int],
    ) -> aura.PandasVector[float]:
        return aura.PandasVector[float](prices * quantities)
    
    con = ai_lake.connect(
            aura_endpoint=os.getenv("aura_endpoint"),
            aura_endpoint_name=os.getenv("aura_endpoint_name"),
            aura_workspace_id=os.getenv("aura_workspace_id"),
            lf_catalog_name=os.getenv("lf_catalog_name"),
            access_key=os.getenv("access_key"),
            secret_key=os.getenv("secret_key"),
            default_database=target_database,
            use_single_cn_mode=True
    
    )
    
    # 注册UDF
    conncreate_scalar_function(
            calculate_sum,
            database=target_database
        )

• 对于volatility参数，3个枚举类型的含义是：
  • VolatilityType.VOLATILE：函数结果可能在任何时候都变化。
  • VolatilityType.STABLE：函数对于固定输入其结果在一次扫描里不变。
  • VolatilityType.IMMUTABLE：函数对于相同的输入总是输出相同的结果。

  volatility参数不会影响函数下推执行，对于IMMUTABLE/STABLE/VOLATILE类型的函数，Python UDF函数都可以下推到DN执行。

• 对于packages参数，如果用户没有指定：
  • 注册Pyarrow UDF会自动使用后端环境的pyarrow版本。
  • 注册Pandas UDF会自动使用后端环境的pandas版本。

  为了考虑到安装Python第三方依赖的隔离性，packages参数中指定的包会被传给元戎RuntimeEnv进行安装，默认的安装方式是venv。

• 对于runtime_env参数，用户需按照元戎给定的RuntimeEnv格式（JSON格式）正确给出，请参见元戎RuntimeEnv文档。其中：
  • 如果用户没有指定runtime_env参数，就会把packages参数中指定的包当作RuntimeEnv的venv来安装。
  • RuntimeEnv建议用户只使用venv字段，其他的字段如pip, conda，working_dir等对于UDF暂时不支持。
  • 不允许用户指定venv的name字段，name字段会根据RuntimeEnv的内容进行SHA-1值的生成，并用于后续在元戎配置项中指定的virtual_env_idle_time时间内进行RuntimeEnv复用。

Builtin UDF注册参数

注册Builtin UDF的作用是获得数据库已存在的函数的句柄，无实际注册的操作。

不管是显式注册还是隐式注册，不管是Scalar UDF、UDAF、UDTF，对于注册Builtin类型的UDF，目前都接受用户传入以下参数：

Builtin UDF注册参数注意事项

对于signature参数，目前允许用户传入，用户传入的优先级高于自动推断。当用户不传入时，进入参数/返回值类型自动推断逻辑，详情请参见signature参数的类型推断。

signature参数类型的推断

对于signature参数，允许用户传入参数/返回值类型，也允许用户不传入。

• 如果用户传入signature参数，不需要原始Python函数使用类型注解（type hints）语法，此时可以支持及时操作式的注册UDF。
• 如果用户不传入signature参数，推荐对原始Python函数使用类型注解（type hints）语法，此时不能支持及时操作式的注册UDF。

两种方式的区别总结如下：

此处的及时操作指的是读取–求值–输出循环（Read–Eval–Print Loop, REPL），通常见于Python用户交互式终端（Terminal）中。

类型注解（type hints）语法由Python 3.5引入（PEP 484），在函数定义中，类型注解通过在参数名后加冒号（:）和类型，以及在参数列表末尾使用箭头（->）指定返回类型来实现，示例如下：

def greet(name: str) -> str:
    return f"Hello, {name}"
from typing import List, Dict, Optional

def process_data(data: List[int]) -> Dict[str, Optional[int]]:
    return {"max": max(data) if data else None}

对于Python/Pyarrow/Pandas UDF，注册时要求强数据类型，所有的参数/返回值都需要指定类型。如果用户不能通过原始Python函数的类型注解语法来指明，那么就需要用户主动使用signature参数指定ibis DataType。

对于Builtin UDF，注册时不要求强数据类型（因为数据库中已经注册了该UDF函数）。如果用户不能通过原始Python函数的类型注解语法来指明，那么推荐用户只写出参数名称，不写出类型；如果用户后续用到Builtin UDF的返回值（非Top SELECT UDF），那么需要指明函数返回值类型，必要时需要用户主动使用signature参数指定ibis DataType，如果不需要（Top SELECT UDF）则允许用户不写出函数返回值类型。

对于用户不传入signature参数，依赖自动推断时的总结如下：

对于用户不传入signature参数，自动推断的情况，底层实现原理是inspect.signature。目前，接受用户传入以下参数/返回值类型：

注意事项：

• Python内置int类型属于numbers.Integral的子类。
• Python内置float类型属于numbers.Real的子类。

上表中没有列出的Python类型，都是目前暂时不支持的自动转化类型。

对于用户不传入signature参数，同时也没有写出Python类型注解（type hints）语法的参数/返回值，目前自动推断采取如下的方式处理：

inspect.signature对于参数类型（Parameter.kind）的分类如下：

UDF直接操作语法

在注册和使用分离的场景下，为使用者提供Scalar UDF、UDAF、UDTF的直接操作语法，使用者只需要知道UDF名称（name）、所在的数据库名称（database），就可以直接操作使用UDF。以下操作依赖Backend会话对象的udf属性进行。

signature(name, database=None)

描述：从后端DB中返回UDF的函数签名和返回值类型。

输入参数：

• name（str）- UDF函数名称。
• database（str）- UDF函数所属的数据库名称。

返回值类型：aura_frame.ibis.common.annotations.Signature - 已注册的UDF函数签名和返回值类型。

get(name, database=None)

描述：从后端DB中返回UDF函数。

输入参数：

• name（str）- UDF函数名称。
• database（str）- UDF函数所属的数据库名称。

返回值类型：Callable[..., ibis.expr.types.Value] - 已注册的UDF函数。

names(database=None)

描述：从后端DB中返回所有的UDF函数名称。

输入参数：

• database（str）：UDF函数所属的数据库名称。

返回值类型：List[str] - 所有已注册的UDF函数名称。

unregister(name, database=None)

描述：从后端DB中删除指定的UDF函数。

输入参数：

• database（str）：UDF函数所属的数据库名称。

返回值类型：None

UDF WITH ARGUMENTS使用语法

不管是使用UDF显式注册语法，UDF隐式注册语法后返回的UDF算子，还是通过UDF直接操作语法返回的已注册UDF，目前都支持用户通过with_arguments方法传入参数，参数总体来说分为两类：

• 特殊含义的参数名，这些参数用来配置UDF运行时的资源、并发和执行上限时间，例如concurrency、timeout、dpu、apu，具体情况可见UDF运行时配置列表。所有的UDF类型目前都支持用户使用with_arguments传入配置参数。
• 一般的参数名，这些参数用来配置UDF初始化时的状态，只有Class UDF、Class UDTF、UDAF这些用户在__init__方法中定义了可选参数的情况下可以传入Scalar标量值。实现一次__init__初始化内部状态，后续多次使用。

所有经过with_arguments方法传入的参数值都是Scalar标量，所有参数值整体作为一个**kwargs（Python字典类型）传入。

create_agg_function(udf, /, input_type=InputType.PYTHON, *, database, name=None, signature=None, imports=None, packages=None, runtime_env=None, comment=None, **kwargs)

描述：注册一个用户自定义聚合函数（UDAF）。

输入参数：

• udf (Callable) ：要注册的用户自定义聚合函数。
• input_type (InputType) ：函数的输入类型，可选值包括PYTHON、BUILTIN、PYARROW、PANDAS。
• database (str) ：函数所属的数据库名称。
• name (Optional[str]) ：函数的名称，默认为None。
• signature (Optional[Signature]) ：函数的签名定义，可选。
• imports (Optional[Union[str, Iterable[str]]]) ：函数所需的导入模块，可以是单个字符串或字符串列表。
• packages (Optional[Union[str, Iterable[str]]]) ：函数所需的依赖包，可以是单个包名或包名列表。
• runtime_env (Optional[Union[str, Mapping[str, Any]]])：函数所需的依赖包和安装方法，可以是JSON格式的字符串或者Python字典，可选字段参见元戎RuntimeEnv文档。
• comment (Optional[str]) ：函数的注释说明，可选。
• **kwargs：其他可选的关键字参数。

返回值类型：

Callable ：注册后的函数。

create_scalar_function(udf, /, input_type=InputType.PYTHON, *, database, name=None, imports=None, packages=None, runtime_env=None, comment=None, **kwargs)

描述：注册一个新的标量用户自定义函数（Scalar UDF）。

输入参数：

• udf (Callable) ：要注册的用户自定义标量函数。
• input_type (InputType) ：函数的输入类型，可选值包括PYTHON、BUILTIN、PYARROW、PANDAS。
• database (str) ：函数所属的数据库名称。
• name (Optional[str]) ：函数的名称，默认为None。
• imports (Optional[Union[str, Iterable[str]]]) ：函数所需的导入模块，可以是单个字符串或字符串列表。
• packages (Optional[Union[str, Iterable[str]]]) ：函数所需的依赖包，可以是单个包名或包名列表。
• runtime_env (Optional[Union[str, Mapping[str, Any]]])：函数所需的依赖包和安装方法，可以是JSON格式的字符串或者Python字典，可选字段参见元戎RuntimeEnv文档。
• comment (Optional[str]) ：函数的注释说明，可选。
• **kwargs：其他可选的关键字参数。

返回值类型：tuple[str, ...]。

create_table_function(udf, /, input_type=InputType.PYTHON, *, database, name=None, signature=None, imports=None, packages=None, runtime_env=None, comment=None, **kwargs)

描述：注册一个新的用户自定义表值函数（UDTF）。

输入参数：

• udf (Callable) ：要注册的用户自定义表值函数。
• input_type (InputType) ：函数的输入类型，可选值包括PYTHON、BUILTIN。
• database (str) ：函数所属的数据库名称。
• name (Optional[str]) ：函数的名称，默认为None。
• signature (Optional[Signature]) ：函数的签名定义，可选。
• imports (Optional[Union[str, Iterable[str]]]) ：函数所需的导入模块，可以是单个字符串或字符串列表。
• packages (Optional[Union[str, Iterable[str]]]) ：函数所需的依赖包，可以是单个包名或包名列表。
• runtime_env (Optional[Union[str, Mapping[str, Any]]])：函数所需的依赖包和安装方法，可以是JSON格式的字符串或者Python字典，可选字段参见元戎RuntimeEnv文档。
• comment (Optional[str]) ：函数的注释说明，可选。
• **kwargs：其他可选的关键字参数。

返回值类型 ：

Callable ：注册后的函数。

delete_function(name, /, *, database=None, if_it_exists=False)

描述：从指定数据库中删除一个用户自定义函数。

输入参数：

• name (str) ：要删除的函数名称。
• database (Optional[str]) ：要删除函数的数据库名称。如果未指定，则使用默认数据库。
• if_exists (bool) ：如果为True，则仅在函数存在时执行删除操作，避免因函数不存在而引发错误。

返回值类型：str。

describe_function(name, /, *, database=None)

描述：描述指定数据库中的函数。

输入参数：

• name (str) ：函数名称。
• database (Optional[str]) ：函数所在的数据库名称。

返回值类型：str，函数的元数据信息。

get_function(name, *, database=None)

描述：从指定或默认数据库中获取一个用户自定义函数。

输入参数：

• func_name (str) ：要获取的用户自定义函数的名称。
• database (Optional[str]) ：要获取函数的数据库名称。如果未提供，则使用默认数据库。

返回值类型 ：Callable，可调用的函数。

list_functions(database=None)

描述：列出指定数据库或默认数据库中的所有用户自定义函数。

输入参数：

database (Optional[str]) ：要列出函数的数据库名称。如果未提供，则使用默认数据库。

返回值类型：pandas.DataFrame。