feat: implement udf server in databend

.github/workflows/reuse.linux.yml

@@ -185,6 +185,33 @@ jobs:
         with:
           name: test-sqllogic-standalone-${{ matrix.dirs }}
 
+  sqllogic_standalone_udf_server:
+    name: sqllogic_standalone_${{ matrix.dirs }}
+    runs-on: [self-hosted, X64, Linux, 4c8g, "${{ inputs.runner_provider }}"]
+    needs: [build, check]
+    strategy:
+      matrix:
+        dirs:
+          - "udf_server"
+    steps:
+      - uses: actions/checkout@v3
+      - name: Start UDF Server
+        run: |
+          pip install pyarrow
+          python3 tests/udf-server/udf_test.py &
+          sleep 2
+      - uses: ./.github/actions/test_sqllogic_standalone_linux
+        timeout-minutes: 15
+        with:
+          dirs: ${{ matrix.dirs }}
+          handlers: mysql,http,clickhouse
+          storage-format: all
+      - name: Upload failure
+        if: failure() || cancelled()
+        uses: ./.github/actions/artifact_failure
+        with:
+          name: test-sqllogic-standalone-${{ matrix.dirs }}
+
   sqllogic_standalone_with_native:
     name: sqllogic_standalone_${{ matrix.dirs }}_${{ matrix.format }}
     runs-on: [self-hosted, X64, Linux, 4c8g, "${{ inputs.runner_provider }}"]

scripts/ci/deploy/config/databend-query-node-1.toml

@@ -38,6 +38,9 @@ table_engine_memory_enabled = true
 default_storage_format = 'parquet'
 default_compression = 'zstd'
 
+enable_udf_server = true
+udf_server_allow_list = ['http://0.0.0.0:8815']
+
 [[query.users]]
 name = "root"
 auth_type = "no_password"

scripts/ci/deploy/config/databend-query-node-native.toml

@@ -32,6 +32,9 @@ cluster_id = "test_cluster"
 
 table_engine_memory_enabled = true
 
+enable_udf_server = true
+udf_server_allow_list = ['http://0.0.0.0:8815']
+
 [[query.users]]
 name = "root"
 auth_type = "no_password"

src/common/exception/src/exception_code.rs

@@ -271,6 +271,10 @@ build_exceptions! {
     IllegalUDFFormat(2601),
     UnknownUDF(2602),
     UdfAlreadyExists(2603),
+    UDFServerConnectError(2604),
+    UDFSchemaMismatch(2605),
+    UnsupportedDataType(2606),
+    UDFDataError(2607),
 
     // Database error codes.
     UnknownDatabaseEngine(2701),

src/meta/app/src/principal/mod.rs

@@ -38,6 +38,9 @@ pub use user_auth::AuthInfo;
 pub use user_auth::AuthType;
 pub use user_auth::PasswordHashMethod;
 pub use user_defined_file_format::UserDefinedFileFormat;
+pub use user_defined_function::LambdaUDF;
+pub use user_defined_function::UDFDefinition;
+pub use user_defined_function::UDFServer;
 pub use user_defined_function::UserDefinedFunction;
 pub use user_grant::GrantEntry;
 pub use user_grant::GrantObject;

src/meta/app/src/principal/user_defined_function.rs

@@ -12,44 +12,114 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::convert::TryFrom;
+use std::fmt::Display;
+use std::fmt::Formatter;
 
-use common_exception::ErrorCode;
-use common_exception::Result;
-use serde::Deserialize;
-use serde::Serialize;
+use common_expression::types::DataType;
 
-#[derive(Serialize, Deserialize, Clone, Debug, Eq, PartialEq, Default)]
-#[serde(default)]
-pub struct UserDefinedFunction {
-    pub name: String,
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct LambdaUDF {
     pub parameters: Vec<String>,
+    pub definition: String,
+}
 
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct UDFServer {
+    pub address: String,
+    pub handler: String,
+    pub language: String,
+    pub arg_types: Vec<DataType>,
+    pub return_type: DataType,
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum UDFDefinition {
+    LambdaUDF(LambdaUDF),
+    UDFServer(UDFServer),
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct UserDefinedFunction {
+    pub name: String,
     pub description: String,
-    pub definition: String,
+    pub definition: UDFDefinition,
 }
 
 impl UserDefinedFunction {
-    pub fn new(name: &str, parameters: Vec<String>, definition: &str, description: &str) -> Self {
+    pub fn create_lambda_udf(
+        name: &str,
+        parameters: Vec<String>,
+        definition: &str,
+        description: &str,
+    ) -> Self {
         Self {
             name: name.to_string(),
-            parameters,
-            definition: definition.to_string(),
             description: description.to_string(),
+            definition: UDFDefinition::LambdaUDF(LambdaUDF {
+                parameters,
+                definition: definition.to_string(),
+            }),
         }
     }
-}
 
-impl TryFrom<Vec<u8>> for UserDefinedFunction {
-    type Error = ErrorCode;
+    pub fn create_udf_server(
+        name: &str,
+        address: &str,
+        handler: &str,
+        language: &str,
+        arg_types: Vec<DataType>,
+        return_type: DataType,
+        description: &str,
+    ) -> Self {
+        Self {
+            name: name.to_string(),
+            description: description.to_string(),
+            definition: UDFDefinition::UDFServer(UDFServer {
+                address: address.to_string(),
+                handler: handler.to_string(),
+                language: language.to_string(),
+                arg_types,
+                return_type,
+            }),
+        }
+    }
+}
 
-    fn try_from(value: Vec<u8>) -> Result<Self> {
-        match serde_json::from_slice(&value) {
-            Ok(udf) => Ok(udf),
-            Err(serialize_error) => Err(ErrorCode::IllegalUDFFormat(format!(
-                "Cannot deserialize user defined function from bytes. cause {}",
-                serialize_error
-            ))),
+impl Display for UDFDefinition {
+    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
+        write!(f, " (")?;
+        match self {
+            UDFDefinition::LambdaUDF(LambdaUDF {
+                parameters,
+                definition,
+            }) => {
+                for (i, item) in parameters.iter().enumerate() {
+                    if i > 0 {
+                        write!(f, ", ")?;
+                    }
+                    write!(f, "{item}")?;
+                }
+                write!(f, ") -> {definition}")?;
+            }
+            UDFDefinition::UDFServer(UDFServer {
+                address,
+                arg_types,
+                return_type,
+                handler,
+                language,
+            }) => {
+                for (i, item) in arg_types.iter().enumerate() {
+                    if i > 0 {
+                        write!(f, ", ")?;
+                    }
+                    write!(f, "{item}")?;
+                }
+                write!(
+                    f,
+                    ") RETURNS {return_type} LANGUAGE {language} HANDLER = {handler} ADDRESS = {address}"
+                )?;
+            }
         }
+        Ok(())
     }
 }

src/meta/app/tests/it/main.rs

@@ -13,7 +13,6 @@
 //  limitations under the License.
 
 mod file_format;
-mod user_defined_function;
 mod user_grant;
 mod user_info;
 mod user_privilege;

src/meta/proto-conv/src/lib.rs

@@ -76,6 +76,7 @@ mod schema_from_to_protobuf_impl;
 mod share_from_to_protobuf_impl;
 mod stage_from_to_protobuf_impl;
 mod table_from_to_protobuf_impl;
+mod udf_from_to_protobuf_impl;
 mod user_from_to_protobuf_impl;
 mod util;
 mod virtual_column_from_to_protobuf_impl;