feat: run numeric WASM UDFs in SQL pipelines

dozer-sql/expression/Cargo.toml

@@ -19,12 +19,14 @@ jsonpath = { path = "../jsonpath" }
 bincode = { workspace = true }
 tokio = "1.34.0"
 async-recursion = "1.0.5"
+wasmi = "=0.31.2"
 
 dozer-deno = { path = "../../dozer-deno", optional = true }
 deno_core = { workspace = true, optional = true }
 
 [dev-dependencies]
 proptest = "1.2.0"
+wat = "=1.0.85"
 
 [features]
 bigdecimal = ["dep:bigdecimal", "sqlparser/bigdecimal"]

dozer-sql/expression/src/builder.rs

@@ -531,9 +531,41 @@ impl ExpressionBuilder {
         }
 
         // config check for udfs
-        let udf_type = udfs.iter().find(|udf| udf.name == function_name);
+        let udf_type = udfs.iter().find(|udf| {
+            udf.name == function_name
+                || (matches!(udf.config, UdfType::Wasm(_))
+                    && udf.name.eq_ignore_ascii_case(&function_name))
+        });
         if let Some(udf_type) = udf_type {
             return match &udf_type.config {
+                UdfType::Wasm(config) => {
+                    if sql_function.over.is_some()
+                        || sql_function.distinct
+                        || sql_function.special
+                        || !sql_function.order_by.is_empty()
+                        || sql_function.return_type.is_some()
+                    {
+                        return Err(Error::Wasm(format!(
+                            "{function_name}: expected a scalar call without SQL function modifiers"
+                        )));
+                    }
+                    let mut args = Vec::new();
+                    for argument in &sql_function.args {
+                        if !matches!(argument, FunctionArg::Unnamed(FunctionArgExpr::Expr(_))) {
+                            return Err(Error::UnsupportedFunctionArg(argument.clone()));
+                        }
+                        args.push(
+                            self.parse_sql_function_arg(false, argument, schema, udfs)
+                                .await?,
+                        );
+                    }
+                    Ok(Expression::WasmUdf(crate::wasm::Udf::new(
+                        udf_type.name.clone(),
+                        config,
+                        args,
+                        schema,
+                    )?))
+                }
                 UdfType::Onnx(config) => {
                     #[cfg(feature = "onnx")]
                     {

dozer-sql/expression/src/error.rs

@@ -12,6 +12,8 @@ use crate::{aggregate::AggregateFunctionType, operator::BinaryOperatorType};
 
 #[derive(Debug, Error)]
 pub enum Error {
+    #[error("WASM UDF error: {0}")]
+    Wasm(String),
     #[error("Unsupported SQL expression: {0:?}")]
     UnsupportedExpression(Expr),
     #[error("Unsupported SQL function arg: {0:?}")]

dozer-sql/expression/src/execution.rs

@@ -20,6 +20,7 @@ use dozer_types::types::{Field, FieldType, Schema, SourceDefinition};
 
 #[derive(Clone, Debug, PartialEq)]
 pub enum Expression {
+    WasmUdf(crate::wasm::Udf),
     Column {
         index: usize,
     },
@@ -110,6 +111,7 @@ pub enum Expression {
 impl Expression {
     pub fn to_string(&self, schema: &Schema) -> String {
         match &self {
+            Expression::WasmUdf(udf) => udf.to_string(schema),
             Expression::Column { index } => schema.fields[*index].name.clone(),
             Expression::Literal(value) => format!("{}", value),
             Expression::UnaryOperator { operator, arg } => {
@@ -317,6 +319,7 @@ impl ExpressionType {
 impl Expression {
     pub fn evaluate(&mut self, record: &Record, schema: &Schema) -> Result<Field, Error> {
         match self {
+            Expression::WasmUdf(udf) => udf.evaluate(record, schema),
             Expression::Literal(field) => Ok(field.clone()),
             Expression::Column { index } => Ok(record.values[*index].clone()),
             Expression::BinaryOperator {
@@ -383,6 +386,7 @@ impl Expression {
 
     pub fn get_type(&self, schema: &Schema) -> Result<ExpressionType, Error> {
         match self {
+            Expression::WasmUdf(udf) => Ok(udf.get_type()),
             Expression::Literal(field) => {
                 let field_type = field.ty();
                 match field_type {

dozer-sql/expression/src/lib.rs

@@ -16,6 +16,7 @@ mod logical;
 mod mathematical;
 pub mod operator;
 pub mod scalar;
+pub mod wasm;
 
 #[cfg(feature = "javascript")]
 mod javascript;