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
  
  con = ibis.fabric.connect(...)
  
  # 注册UDAF
  udf = con.udaf.python.register(
      outer(), #从外部引入的fn
      imports=["process.py"] #为fn添加文件路径
  )

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

    import fabric_data as fabric
    import pyarrow as pa
    import pyarrow.compute as pc
    
    # === 用户传入signature，需要依赖PyarrowVector ===
    def calculate_sum(
        prices: pa.ChunkedArray,
        quantities: pa.ChunkedArray,
    ) -> pa.ChunkedArray:
        return pc.multiply(prices, quantities)
    
    con = ibis.fabric.connect(...)
    
    # 注册UDF
    udf = con.udf.pyarrow.register(
        fn=calculate_sum
        signature=fabric.Signature(
            parameters=[
                fabric.Parameter(name="price", annotation=fabric.PyarrowVector[float]),
                fabric.Parameter(name="quantity", annotation=fabric.PyarrowVector[int]),
            ],
            return_annotation=fabric.PyarrowVector[float],
        ),
    )
    
    # === 用户不传入signature，也需要依赖PyarrowVector ===
    def calculate_sum(
        prices: fabric.PyarrowVector[float],
        quantities: fabric.PyarrowVector[int],
    ) -> fabric.PyarrowVector[float]:
        return fabric.PyarrowVector[float](pc.multiply(prices, quantities))
    
    con = ibis.fabric.connect(...)
    
    # 注册UDF
    udf = con.udf.pyarrow.register(
        fn=calculate_sum
    )

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

    import fabric_data as fabric
    import pandas as pd
    
    # === 用户传入signature，需要依赖PandasVector ===
    def calculate_sum(
        prices: pd.Series, 
        quantities: pd.Series,
    ) -> pd.Series:
        return pd.Series(prices * quantities, dtype=pd.Float64Dtype())
    
    con = ibis.fabric.connect(...)
    
    # 注册UDF
    udf = con.udf.pandas.register(
        fn=calculate_sum
        signature=fabric.Signature(
            parameters=[
                fabric.Parameter(name="price", annotation=fabric.PandasVector[float]),
                fabric.Parameter(name="quantity", annotation=fabric.PandasVector[int]),
            ],
            return_annotation=fabric.PandasVector[float],
        ),
    )
    
    # === 用户不传入signature，也需要依赖PandasVector ===
    def calculate_sum(
        prices: fabric.PandasVector[float], 
        quantities: fabric.PandasVector[int],
    ) -> fabric.PandasVector[float]:
        return fabric.PandasVector[float](prices * quantities, dtype=pd.Float64Dtype())
    
    con = ibis.fabric.connect(...)
    
    # 注册UDF
    udf = con.udf.pandas.register(
        fn=calculate_sum
    )

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

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

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

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函数所属的数据库名称。

返回值类型：fabric_data.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字典类型）传入。