RFC: Implement Databend UDF Server

[gitccl]

Summary

Implement Databend UDF Server.

Motivation

UDF Server is a type of UDF, which can run independently on any server and be accessed by databases through network connections. The security and isolation of UDF Server are relatively high, because problems with UDF Server will not affect the normal execution of the database.

At present, Databend has not yet implemented UDF Server, and only provides lambda UDF support. Therefore, we intend to implement Databend UDF Server, allowing users to create, delete and call UDF. After it is implemented, our users will be able to use UDF conveniently, further improving the flexibility and scalability of Databend, and making data analysis and processing more efficient and convenient.

Guide-level explanation

You can define your own functions and call these functions in Databend.

1. define your functions and start the UDF server (using python as example)

   from databend.udf import udf, UdfServer
   
   # Define a scalar function
   @udf(input_types=['INT', 'INT'], result_type='INT')
   def plus_int(x, y):
       return x + y
   
   # Start a UDF server
   if __name__ == '__main__':
       server = UdfServer(location="0.0.0.0:8888")
       server.add_function(plus_int)
       server.serve()

2. Add the UDF Server into Databend

   CREATE FUNCTION plus_int AS (int, int) -> int address 'http://localhost:8888'

3. Call these functions in Databend

   select plus_int(1, 2);
   ---------------
   3
   

Reference-level explanation

The management of UDF Server

Extending the syntax and functionality of the CREATE FUNCTION to support creating a UDF Server. Extend the ALTER FUNCTION and DELETE FUNCTION to support UDF Server modification and deletion.

The syntax of CREATE FUNCTION command for creating UDF Server is as follows:

CREATE FUNCTION [ IF NOT EXISTS ] function_name AS ( [argument_type] ) ->
return_type address 'udf_server_address'

Like lambda UDF, the information of UDF Server also need to be stored in meta. The related struct used to store UDF information is defined as follows:

pub struct LambdaUDF {
    pub parameters: Vec<String>,
    pub definition: String,
}

pub struct UDFServer {
    pub address: String,
    pub arg_types: Vec<DataType>,
    pub return_type: DataType,
}

pub enum UDFDefinition {
    LambdaUDF(LambdaUDF),
    UDFServer(UDFServer),
}

pub struct UserDefinedFunction {
    pub name: String,
    pub description: String,
    pub definition: UDFDefinition,
}

The resolution and execution of UDF Server

The call of UDF Server is very similar to the normal function call. In the bind stage, we will resolve the call to UDF Server into a ScalarExpr::UDFServerCall, whose structure is defined as follows:

pub struct UDFServerCall {
    pub span: Span,
    pub func_name: String,
    pub server_addr: String,
    pub return_type: DataType,
    pub arguments: Vec<ScalarExpr>,
}

There are some other related exprs, we can also add a new enum to them:

pub enum RawExpr<Index: ColumnIndex = usize> {
    UDFServerCall {
        span: Span,
        func_name: String,
        server_addr: String,
        return_type: DataType,
        arguments: Vec<RawExpr<Index>>,
    }
}

pub enum RemoteExpr<Index: ColumnIndex = usize> {
    UDFServerCall {
        span: Span,
        func_name: String,
        server_addr: String,
        return_type: DataType,
        arguments: Vec<RemoteExpr<Index>>,
    }
}

pub enum Expr<Index: ColumnIndex = usize> {
    UDFServerCall {
        span: Span,
        func_name: String,
        server_addr: String,
        return_type: DataType,
        arguments: Vec<Expr<Index>>,
    }
}

In the evaluate stage, packing the parameters required by this function into a RecordBatch , and send the RecordBatch to the UDF server through Arrow Flight DoExchange RPC, and wait for the UDF server to return the calculation result.

Language-specific UDF Server(using python as example)

Python UDF Server runs as a Arrow Flight Server, it receives the request from the Databend, parses parameters and call the corresponding function to perform the calculation, then return the result of the calculation. Python UDF Server should implement two rpc: GetFlightInfo and DoExchange, used to return the input and output types of the function and call the user defined function to perform calculations.

We will provide users with a Python SDK to help them write UDF Server.

Unresolved questions

• The evaluate of UDF is synchronous, should it be evaluated asynchronously?
• Is UDF deterministic?

Future possibilities

Support user defined table function like:

select * from udtf();

Reference

RisingWave UDF