Feature: Implement Databend UDF Server (OSPP)

[xudong963]

Summary

1. Writing an RFC for the UDF server.
2. Implementing an MVP that provides the functionality to create, delete, and execute UDFs, and writing relevant tests.
3. Creating an example using Python.

Mentor @xudong963

[manulpatel]

Hello @xudong963! I would like to implement this feature as a part of OSPP. Could you please give some hints on any specifics to be included in the proposal?

[Jacob953]

Based on the interpretation of bilibili, this proposal could be divided into 3 parts.

Modify databend kernel:

• Extend the "create function" syntax to support Python UDF Server.
• Make data transfer between Databend kernel and Python UDF Server using Arrow Flight.
• Realize the communication between Eval Expression and Python Server to obtain calculation results.

Implement Python UDF Server:

• Design relevant information such as the address and port of the Python UDF Server.
• Implement the registration mechanism to recognize and call functions in the Python UDF Server.
• Design and implement the mechanism to start the Python UDF Server.

Also, design and execute test cases to ensure that the functions and performance of the modified Databend kernel and Python UDF Server meet the expected requirements.

[gitccl]

I plan to implement Databend UDF Server from the following three aspects:

1. Implement Python UDF Server:

• Arrow Flight is used to communicate between Databend and Python UDF Server.
• Implement the Python UDF Server to receive the request from the Databend and call the registered function to perform the calculation, then return the result of the calculation.
• Provide users with a Python SDK, so that users can easily write Python UDF Server. Users only need to provide the Server address, register custom functions, and start the Server to complete a Python UDF Server. Like following:

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

2. Implement the management of UDF Server in Databend:

• Extend the syntax and functionality of the CREATE FUNCTION to support creating a UDF Server and storing the information of the UDF Server into the meta.
• Extend the ALTER FUNCTION and DELETE FUNCTION to enable Datebend to support UDF Server modification and deletion.

3. Implement the execution of UDF Server in Databend

• The call to UDF Server is resolved into a ScalarExpr::UDFServerCall in the Binder phase, where the UDFServerCall struct is defined as follows:

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

• Implement the Databend kernel to call the Python UDF Server and obtain the execution result in the Evaluator execution phase.