From 4b6244581577b90d2241cba9fc97d5d019b75131 Mon Sep 17 00:00:00 2001 From: Gavin-Yau <2695188238@qq.com> Date: Sat, 26 Sep 2026 21:01:10 +0800 Subject: [PATCH] feat: run numeric WASM UDFs in SQL pipelines --- Cargo.lock | 91 ++++ dozer-sql/expression/Cargo.toml | 2 + dozer-sql/expression/src/builder.rs | 34 +- dozer-sql/expression/src/error.rs | 2 + dozer-sql/expression/src/execution.rs | 4 + dozer-sql/expression/src/lib.rs | 1 + dozer-sql/expression/src/wasm.rs | 365 ++++++++++++++++ dozer-sql/expression/src/wasm/tests.rs | 232 ++++++++++ dozer-sql/src/builder/mod.rs | 1 + dozer-sql/src/tests/mod.rs | 1 + dozer-sql/src/tests/wasm_udf.rs | 399 ++++++++++++++++++ dozer-types/src/models/config.rs | 7 +- dozer-types/src/models/udf_config.rs | 129 ++++++ dozer-types/src/tests/udf_yaml_deserialize.rs | 55 +++ examples/wasm-udf/.gitignore | 1 + examples/wasm-udf/README.md | 120 ++++++ examples/wasm-udf/assembly/index.ts | 18 + examples/wasm-udf/dozer-config.yaml | 6 + examples/wasm-udf/package-lock.json | 54 +++ examples/wasm-udf/package.json | 10 + examples/wasm-udf/udfs/wasm/my_function.wasm | Bin 0 -> 319 bytes examples/wasm-udf/udfs/wasm/my_function.wat | 53 +++ json_schemas/dozer.json | 85 +++- 23 files changed, 1664 insertions(+), 6 deletions(-) create mode 100644 dozer-sql/expression/src/wasm.rs create mode 100644 dozer-sql/expression/src/wasm/tests.rs create mode 100644 dozer-sql/src/tests/wasm_udf.rs create mode 100644 examples/wasm-udf/.gitignore create mode 100644 examples/wasm-udf/README.md create mode 100644 examples/wasm-udf/assembly/index.ts create mode 100644 examples/wasm-udf/dozer-config.yaml create mode 100644 examples/wasm-udf/package-lock.json create mode 100644 examples/wasm-udf/package.json create mode 100644 examples/wasm-udf/udfs/wasm/my_function.wasm create mode 100644 examples/wasm-udf/udfs/wasm/my_function.wat diff --git a/Cargo.lock b/Cargo.lock index 55a96a2c33..d7583df766 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2768,6 +2768,12 @@ version = "0.3.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fea41bba32d969b513997752735605054bc0dfa92b4c56bf1189f2e174be7a10" +[[package]] +name = "downcast-rs" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "75b325c5dbd37f80359721ad39aca5a29fb04c89279657cffdda8736d0c0b9d2" + [[package]] name = "dozer-cli" version = "0.4.0" @@ -3069,6 +3075,8 @@ dependencies = [ "proptest", "sqlparser 0.35.0", "tokio", + "wasmi", + "wat", ] [[package]] @@ -4535,6 +4543,12 @@ dependencies = [ "serde", ] +[[package]] +name = "indexmap-nostd" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e04e2fd2b8188ea827b32ef11de88377086d690286ab35747ef7f9bf3ccb590" + [[package]] name = "indicatif" version = "0.17.8" @@ -4808,6 +4822,12 @@ version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "830d08ce1d1d941e6b30645f1a0eb5643013d835ce3779a5fc208261dbe10f55" +[[package]] +name = "leb128" +version = "0.2.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c83bff1d572d6b9aeef67ddfc8448e4a3737909cb28e81f97c791b9018703e52" + [[package]] name = "lexical-core" version = "0.8.5" @@ -9870,6 +9890,15 @@ version = "0.2.92" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "af190c94f2773fdb3729c55b007a722abb5384da03bc0986df4c289bf5567e96" +[[package]] +name = "wasm-encoder" +version = "0.41.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "972f97a5d8318f908dded23594188a90bcd09365986b1163e66d70170e5287ae" +dependencies = [ + "leb128", +] + [[package]] name = "wasm-streams" version = "0.3.0" @@ -9883,6 +9912,68 @@ dependencies = [ "web-sys", ] +[[package]] +name = "wasmi" +version = "0.31.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77a8281d1d660cdf54c76a3efa9ddd0c270cada1383a995db3ccb43d166456c7" +dependencies = [ + "smallvec", + "spin 0.9.8", + "wasmi_arena", + "wasmi_core", + "wasmparser-nostd", +] + +[[package]] +name = "wasmi_arena" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "104a7f73be44570cac297b3035d76b169d6599637631cf37a1703326a0727073" + +[[package]] +name = "wasmi_core" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dcf1a7db34bff95b85c261002720c00c3a6168256dcb93041d3fa2054d19856a" +dependencies = [ + "downcast-rs", + "libm", + "num-traits", + "paste", +] + +[[package]] +name = "wasmparser-nostd" +version = "0.100.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d5a015fe95f3504a94bb1462c717aae75253e39b9dd6c3fb1062c934535c64aa" +dependencies = [ + "indexmap-nostd", +] + +[[package]] +name = "wast" +version = "70.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a3d5061300042ff5065123dae1e27d00c03f567d34a2937c8472255148a216dc" +dependencies = [ + "bumpalo", + "leb128", + "memchr", + "unicode-width", + "wasm-encoder", +] + +[[package]] +name = "wat" +version = "1.0.85" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "afd7357b6cc46d46a2509c43dcb1dd4131dafbf4e75562d87017b5a05ffad2d6" +dependencies = [ + "wast", +] + [[package]] name = "web-sys" version = "0.3.69" diff --git a/dozer-sql/expression/Cargo.toml b/dozer-sql/expression/Cargo.toml index 3423d7904a..eb1a6b07b2 100644 --- a/dozer-sql/expression/Cargo.toml +++ b/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"] diff --git a/dozer-sql/expression/src/builder.rs b/dozer-sql/expression/src/builder.rs index cefcda0039..c5b8d917c6 100644 --- a/dozer-sql/expression/src/builder.rs +++ b/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")] { diff --git a/dozer-sql/expression/src/error.rs b/dozer-sql/expression/src/error.rs index 7078229713..2287000436 100644 --- a/dozer-sql/expression/src/error.rs +++ b/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:?}")] diff --git a/dozer-sql/expression/src/execution.rs b/dozer-sql/expression/src/execution.rs index 1c2891b234..9813154f77 100644 --- a/dozer-sql/expression/src/execution.rs +++ b/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 { 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 { match self { + Expression::WasmUdf(udf) => Ok(udf.get_type()), Expression::Literal(field) => { let field_type = field.ty(); match field_type { diff --git a/dozer-sql/expression/src/lib.rs b/dozer-sql/expression/src/lib.rs index c94463f9d9..b61d4858fa 100644 --- a/dozer-sql/expression/src/lib.rs +++ b/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; diff --git a/dozer-sql/expression/src/wasm.rs b/dozer-sql/expression/src/wasm.rs new file mode 100644 index 0000000000..b1f3da59e3 --- /dev/null +++ b/dozer-sql/expression/src/wasm.rs @@ -0,0 +1,365 @@ +//! Sandboxed, fixed-arity numeric WebAssembly scalar functions. +use std::{fs::File, io::Read, path::PathBuf, sync::Arc}; + +use dozer_types::{ + models::udf_config::{UdfConfig, UdfType, WasmConfig}, + types::{Field, FieldType, Record, Schema, SourceDefinition}, +}; +use wasmi::{ + core::{Trap, ValueType}, + Config, Engine, ExternType, Instance, Linker, Module, Store, StoreLimits, StoreLimitsBuilder, + Value, +}; + +use crate::{ + error::Error, + execution::{Expression, ExpressionType}, +}; + +const MAX_MODULE_BYTES: u64 = 10 * 1024 * 1024; +const MAX_FUEL: u64 = 10_000_000; +const MAX_MEMORY: usize = 64 * 1024 * 1024; + +fn error(name: &str, message: impl std::fmt::Display) -> Error { + Error::Wasm(format!("{name}: {message}")) +} + +#[derive(Debug)] +struct Compiled { + engine: Engine, + module: Module, + bytes: Vec, + config: WasmConfig, + export: String, + params: Vec, + result: ValueType, +} + +impl Compiled { + fn load(name: &str, config: &WasmConfig) -> Result { + if config.fuel == 0 || config.fuel > MAX_FUEL { + return Err(error( + name, + format!("fuel must be between 1 and {MAX_FUEL}"), + )); + } + if config.memory_limit == 0 || config.memory_limit > MAX_MEMORY { + return Err(error( + name, + format!("memory_limit must be between 1 and {MAX_MEMORY} bytes"), + )); + } + let mut path = PathBuf::from(&config.path); + if path.extension().is_none() { + path.set_extension("wasm"); + } + let metadata = std::fs::metadata(&path) + .map_err(|e| error(name, format!("{}: {e}", path.display())))?; + if !metadata.is_file() || metadata.len() > MAX_MODULE_BYTES { + return Err(error( + name, + "module must be a regular file of at most 10 MiB", + )); + } + let file = + File::open(&path).map_err(|e| error(name, format!("{}: {e}", path.display())))?; + let mut bytes = Vec::new(); + file.take(MAX_MODULE_BYTES + 1) + .read_to_end(&mut bytes) + .map_err(|e| error(name, e))?; + if bytes.len() as u64 > MAX_MODULE_BYTES { + return Err(error(name, "module exceeds 10 MiB")); + } + let mut engine_config = Config::default(); + engine_config.consume_fuel(true); + let engine = Engine::new(&engine_config); + let module = Module::new(&engine, &bytes[..]).map_err(|e| error(name, e))?; + for import in module.imports() { + let is_abort = import.module() == "env" + && import.name() == "abort" + && matches!(import.ty(), ExternType::Func(ty) + if ty.params() == [ValueType::I32; 4] && ty.results().is_empty()); + if !is_abort { + return Err(error( + name, + format!("unsupported import {}.{}", import.module(), import.name()), + )); + } + } + let export = config.export.as_deref().unwrap_or(name).to_owned(); + let Some(ExternType::Func(ty)) = module.get_export(&export) else { + return Err(error(name, format!("missing function export {export}"))); + }; + let numeric = |ty: &ValueType| { + matches!( + ty, + ValueType::I32 | ValueType::I64 | ValueType::F32 | ValueType::F64 + ) + }; + if ty.results().len() != 1 + || !ty.params().iter().all(numeric) + || !ty.results().iter().all(numeric) + { + return Err(error( + name, + "expected numeric i32/i64/f32/f64 parameters and exactly one numeric result", + )); + } + let params = ty.params().to_vec(); + let result = ty.results()[0]; + let compiled = Self { + engine, + module, + bytes, + config: config.clone(), + export, + params, + result, + }; + // Validate allocation and start functions before any stream records are accepted. + compiled.instantiate(name)?; + Ok(compiled) + } + + fn instantiate(&self, name: &str) -> Result<(Store, Instance), Error> { + let limits = StoreLimitsBuilder::new() + .memory_size(self.config.memory_limit) + .table_elements(4096) + .memories(1) + .tables(1) + .instances(1) + .trap_on_grow_failure(true) + .build(); + let mut store = Store::new(&self.engine, limits); + store.limiter(|limits| limits); + store + .add_fuel(self.config.fuel) + .map_err(|e| error(name, e))?; + let mut linker = Linker::new(&self.engine); + linker + .func_wrap( + "env", + "abort", + |_message: i32, _file: i32, _line: i32, _column: i32| { + Err::<(), Trap>(Trap::new("AssemblyScript abort")) + }, + ) + .map_err(|e| error(name, e))?; + let instance = linker + .instantiate(&mut store, &self.module) + .and_then(|instance| instance.start(&mut store)) + .map_err(|e| error(name, e))?; + Ok((store, instance)) + } +} + +/// Validate all configured WASM functions, including ones not referenced by SQL. +pub fn validate_udfs(udfs: &[UdfConfig]) -> Result<(), Error> { + for udf in udfs { + if let UdfType::Wasm(config) = &udf.config { + if udfs + .iter() + .filter(|other| other.name.eq_ignore_ascii_case(&udf.name)) + .count() + > 1 + { + return Err(error(&udf.name, "duplicate SQL function name")); + } + let mut chars = udf.name.chars(); + if !chars + .next() + .is_some_and(|c| c.is_ascii_alphabetic() || c == '_') + || !chars.all(|c| c.is_ascii_alphanumeric() || c == '_') + { + return Err(error( + &udf.name, + "SQL function name must be a simple identifier", + )); + } + Compiled::load(&udf.name, config)?; + } + } + Ok(()) +} + +#[derive(Clone, Debug)] +pub struct Udf { + name: String, + args: Vec, + compiled: Arc, + nullable: bool, +} + +impl PartialEq for Udf { + fn eq(&self, other: &Self) -> bool { + self.name == other.name + && self.args == other.args + && self.compiled.config == other.compiled.config + && self.compiled.bytes == other.compiled.bytes + } +} + +impl Udf { + pub fn new( + name: String, + config: &WasmConfig, + args: Vec, + schema: &Schema, + ) -> Result { + let compiled = Arc::new(Compiled::load(&name, config)?); + if args.len() != compiled.params.len() { + return Err(error( + &name, + format!( + "expected {} arguments, got {}", + compiled.params.len(), + args.len() + ), + )); + } + let mut nullable = false; + for (index, (arg, param)) in args.iter().zip(&compiled.params).enumerate() { + if matches!(arg, Expression::Literal(Field::Null)) { + nullable = true; + continue; + } + let ty = arg.get_type(schema)?; + // Some existing computed expressions under-report nullability. + // Preserve exact metadata for fields/literals/our own UDFs, and do + // not advertise a non-null result for a potentially nullable input. + nullable |= ty.nullable + || !matches!( + arg, + Expression::Column { .. } | Expression::Literal(_) | Expression::WasmUdf(_) + ); + let valid = match param { + ValueType::I32 | ValueType::I64 => matches!( + ty.return_type, + FieldType::Int + | FieldType::Int8 + | FieldType::I128 + | FieldType::UInt + | FieldType::U128 + ), + ValueType::F32 | ValueType::F64 => ty.return_type == FieldType::Float, + _ => false, + }; + if !valid { + return Err(error(&name, format!("argument {index} has SQL type {}, expected {param:?}; use an explicit CAST", ty.return_type))); + } + } + Ok(Self { + name, + args, + compiled, + nullable, + }) + } + + pub fn get_type(&self) -> ExpressionType { + let ty = match self.compiled.result { + ValueType::I32 | ValueType::I64 => FieldType::Int, + _ => FieldType::Float, + }; + ExpressionType::new(ty, self.nullable, SourceDefinition::Dynamic, false) + } + + pub fn to_string(&self, schema: &Schema) -> String { + format!( + "{}({})", + self.name, + self.args + .iter() + .map(|arg| arg.to_string(schema)) + .collect::>() + .join(",") + ) + } + + pub fn evaluate(&mut self, record: &Record, schema: &Schema) -> Result { + let fields = self + .args + .iter_mut() + .map(|arg| arg.evaluate(record, schema)) + .collect::, _>>()?; + if fields.iter().any(|field| matches!(field, Field::Null)) { + return Ok(Field::Null); + } + let params = fields + .iter() + .zip(&self.compiled.params) + .enumerate() + .map(|(index, (field, ty))| { + to_value(field, *ty).ok_or_else(|| { + error( + &self.name, + format!("argument {index} is not a finite, in-range {ty:?}"), + ) + }) + }) + .collect::, _>>()?; + // Stores are never shared across rows. Re-evaluating a deleted/updated row + // sees the same initialized globals and memory as its original insertion. + let (mut store, instance) = self.compiled.instantiate(&self.name)?; + let function = instance + .get_func(&store, &self.compiled.export) + .ok_or_else(|| error(&self.name, "function export disappeared"))?; + let mut result = [Value::default(self.compiled.result)]; + function + .call(&mut store, ¶ms, &mut result) + .map_err(|e| error(&self.name, e))?; + match result[0] { + Value::I32(value) => Ok(Field::Int(i64::from(value))), + Value::I64(value) => Ok(Field::Int(value)), + Value::F32(value) => finite_float(&self.name, f64::from(f32::from(value))), + Value::F64(value) => finite_float(&self.name, f64::from(value)), + _ => Err(error(&self.name, "unsupported result type")), + } + } +} + +fn finite_float(name: &str, value: f64) -> Result { + if value.is_finite() { + Ok(Field::Float(value.into())) + } else { + Err(error(name, "result is NaN or infinity")) + } +} + +fn to_value(field: &Field, ty: ValueType) -> Option { + match ty { + ValueType::I32 | ValueType::I64 => { + let value = match field { + Field::Int(v) => i128::from(*v), + Field::Int8(v) => i128::from(*v), + Field::I128(v) => *v, + Field::UInt(v) => i128::from(*v), + Field::U128(v) => i128::try_from(*v).ok()?, + _ => return None, + }; + if ty == ValueType::I32 { + Some(Value::I32(i32::try_from(value).ok()?)) + } else { + Some(Value::I64(i64::try_from(value).ok()?)) + } + } + ValueType::F32 | ValueType::F64 => { + let Field::Float(value) = field else { + return None; + }; + if !value.0.is_finite() { + return None; + } + if ty == ValueType::F32 { + let value = value.0 as f32; + value.is_finite().then(|| Value::F32(value.into())) + } else { + Some(Value::F64(value.0.into())) + } + } + _ => None, + } +} + +#[cfg(test)] +mod tests; diff --git a/dozer-sql/expression/src/wasm/tests.rs b/dozer-sql/expression/src/wasm/tests.rs new file mode 100644 index 0000000000..46d8f1debe --- /dev/null +++ b/dozer-sql/expression/src/wasm/tests.rs @@ -0,0 +1,232 @@ +use super::*; +use std::sync::atomic::{AtomicU64, Ordering}; + +struct Fixture(PathBuf); +impl Fixture { + fn new(wat: &str) -> Self { + static ID: AtomicU64 = AtomicU64::new(0); + let path = std::env::temp_dir().join(format!( + "dozer-wasm-{}-{}.wasm", + std::process::id(), + ID.fetch_add(1, Ordering::Relaxed) + )); + std::fs::write(&path, wat::parse_str(wat).unwrap()).unwrap(); + Self(path) + } + fn config(&self) -> WasmConfig { + WasmConfig::new(self.0.to_str().unwrap().into()) + } + fn udf(&self, fields: Vec) -> Result { + Udf::new( + "f".into(), + &self.config(), + fields.into_iter().map(Expression::Literal).collect(), + &Schema::default(), + ) + } +} +impl Drop for Fixture { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.0); + } +} +fn eval(udf: &mut Udf) -> Result { + udf.evaluate(&Record::new(vec![]), &Schema::default()) +} + +#[test] +fn numeric_abi_checks_signed_and_unsigned_boundaries() { + for (wasm_type, minimum, maximum) in [ + ("i32", i32::MIN as i64, i32::MAX as i64), + ("i64", i64::MIN, i64::MAX), + ] { + let fixture = Fixture::new(&format!( + "(module (func (export \"f\") (param {wasm_type}) (result {wasm_type}) local.get 0))" + )); + for field in [ + Field::Int(minimum), + Field::Int(maximum), + Field::UInt(maximum as u64), + Field::Int8(-2), + ] { + let expected = match field { + Field::UInt(v) => v as i64, + Field::Int(v) => v, + Field::Int8(v) => i64::from(v), + _ => unreachable!(), + }; + assert_eq!( + eval(&mut fixture.udf(vec![field]).unwrap()).unwrap(), + Field::Int(expected) + ); + } + for field in [ + Field::I128(i128::from(minimum) - 1), + Field::U128(maximum as u128 + 1), + Field::UInt(u64::MAX), + Field::U128(u128::MAX), + ] { + assert!(eval(&mut fixture.udf(vec![field]).unwrap()) + .unwrap_err() + .to_string() + .contains("in-range")); + } + assert!(fixture.udf(vec![Field::Float(1.0.into())]).is_err()); + } +} + +#[test] +fn floating_abi_rejects_nonfinite_values_and_f32_overflow() { + for ty in ["f32", "f64"] { + let fixture = Fixture::new(&format!( + "(module (func (export \"f\") (param {ty}) (result {ty}) local.get 0))" + )); + assert_eq!( + eval(&mut fixture.udf(vec![Field::Float((-1.25).into())]).unwrap()).unwrap(), + Field::Float((-1.25).into()) + ); + for value in [f64::NAN, f64::INFINITY, f64::NEG_INFINITY] { + assert!(eval(&mut fixture.udf(vec![Field::Float(value.into())]).unwrap()).is_err()); + } + if ty == "f32" { + assert!(eval(&mut fixture.udf(vec![Field::Float(f64::MAX.into())]).unwrap()).is_err()); + } + assert!(fixture.udf(vec![Field::Int(1)]).is_err()); + for value in ["nan", "inf", "-inf"] { + let output = Fixture::new(&format!( + "(module (func (export \"f\") (result {ty}) {ty}.const {value}))" + )); + assert!(eval(&mut output.udf(vec![]).unwrap()) + .unwrap_err() + .to_string() + .contains("NaN or infinity")); + } + } +} + +#[test] +fn null_is_propagated_without_running_the_function() { + let fixture = + Fixture::new("(module (func (export \"f\") (param i32) (result i32) unreachable))"); + let mut udf = fixture.udf(vec![Field::Null]).unwrap(); + assert!(udf.get_type().nullable); + assert_eq!(eval(&mut udf).unwrap(), Field::Null); +} + +#[test] +fn module_validation_rejects_imports_exports_and_unsupported_signatures() { + for wasm in [ + "(module)", + "(module (memory (export \"f\") 1))", + "(module (func (export \"f\")))", + "(module (func (export \"f\") (result i32 i32) i32.const 1 i32.const 2))", + "(module (func (export \"f\") (param externref) (result i32) i32.const 1))", + "(module (import \"env\" \"seed\" (func)) (func (export \"f\") (result i32) i32.const 1))", + "(module (import \"env\" \"abort\" (func (param i32))) (func (export \"f\") (result i32) i32.const 1))", + "(module (import \"env\" \"memory\" (memory 1)) (func (export \"f\") (result i32) i32.const 1))", + ] { + assert!(Fixture::new(wasm).udf(vec![]).is_err(), "accepted: {wasm}"); + } + let fixture = + Fixture::new("(module (func (export \"f\") (param i32) (result i32) local.get 0))"); + assert!(fixture.udf(vec![]).is_err()); + assert!(fixture + .udf(vec![Field::String("not a pointer".into())]) + .is_err()); +} + +#[test] +fn traps_fuel_and_memory_growth_are_bounded() { + for wasm in [ + "(module (func (export \"f\") (result i32) unreachable))", + "(module (func (export \"f\") (result i32) (loop br 0) i32.const 1))", + "(module (memory 1) (func (export \"f\") (result i32) i32.const 1000 memory.grow))", + "(module (table 1 funcref) (func (export \"f\") (result i32) ref.null func i32.const 5000 table.grow))", + "(module (import \"env\" \"abort\" (func $abort (param i32 i32 i32 i32))) (func (export \"f\") (result i32) i32.const 0 i32.const 0 i32.const 0 i32.const 0 call $abort i32.const 1))", + ] { + let fixture = Fixture::new(wasm); + let mut config = fixture.config(); + config.fuel = 1000; + let mut udf = Udf::new("f".into(), &config, vec![], &Schema::default()).unwrap(); + assert!(eval(&mut udf).is_err(), "accepted: {wasm}"); + assert!(eval(&mut udf).is_err(), "a failed invocation leaked state"); + } +} + +#[test] +fn startup_is_metered_and_memory_limits_apply_before_instantiation() { + for wasm in [ + "(module (func $start (loop br 0)) (start $start) (func (export \"f\") (result i32) i32.const 1))", + "(module (func $start unreachable) (start $start) (func (export \"f\") (result i32) i32.const 1))", + "(module (memory 1000) (func (export \"f\") (result i32) i32.const 1))", + "(module (table 5000 funcref) (func (export \"f\") (result i32) i32.const 1))", + ] { + let fixture = Fixture::new(wasm); + let mut config = fixture.config(); + config.fuel = 1000; + assert!(Compiled::load("f", &config).is_err()); + } +} + +#[test] +fn every_record_gets_fresh_globals_and_memory() { + let fixture = Fixture::new( + "(module (global $counter (mut i64) (i64.const 0)) (memory 1) + (func (export \"f\") (result i64) + global.get $counter i64.const 1 i64.add global.set $counter + i32.const 0 i32.const 0 i64.load i64.const 10 i64.add i64.store + i32.const 0 i64.load global.get $counter i64.add))", + ); + let mut udf = fixture.udf(vec![]).unwrap(); + for _ in 0..3 { + assert_eq!(eval(&mut udf).unwrap(), Field::Int(11)); + } + assert_eq!(eval(&mut udf.clone()).unwrap(), Field::Int(11)); +} + +#[test] +fn wasm_integer_arithmetic_retains_its_defined_wrapping_semantics() { + let fixture = Fixture::new( + "(module (func (export \"f\") (param i32) (result i32) local.get 0 i32.const 1 i32.add))", + ); + assert_eq!( + eval(&mut fixture.udf(vec![Field::Int(i32::MAX.into())]).unwrap()).unwrap(), + Field::Int(i32::MIN.into()) + ); +} + +#[test] +fn invalid_limits_missing_files_and_duplicate_names_fail_startup() { + let fixture = Fixture::new("(module (func (export \"f\") (result i32) i32.const 1))"); + for fuel in [0, MAX_FUEL + 1] { + let mut config = fixture.config(); + config.fuel = fuel; + assert!(Compiled::load("f", &config).is_err()); + } + for memory_limit in [0, MAX_MEMORY + 1] { + let mut config = fixture.config(); + config.memory_limit = memory_limit; + assert!(Compiled::load("f", &config).is_err()); + } + let config = UdfConfig { + name: "f".into(), + config: UdfType::Wasm(fixture.config()), + }; + assert!(validate_udfs(&[config.clone(), config]).is_err()); + assert!(Compiled::load( + "f", + &WasmConfig::new(fixture.0.with_extension("missing").to_str().unwrap().into()) + ) + .is_err()); + std::fs::write(&fixture.0, b"not wasm").unwrap(); + assert!(fixture.udf(vec![]).is_err()); + File::create(&fixture.0) + .unwrap() + .set_len(MAX_MODULE_BYTES + 1) + .unwrap(); + assert!(fixture + .udf(vec![]) + .unwrap_err() + .to_string() + .contains("10 MiB")); +} diff --git a/dozer-sql/src/builder/mod.rs b/dozer-sql/src/builder/mod.rs index 3ccf3ed356..7da5a4bce6 100644 --- a/dozer-sql/src/builder/mod.rs +++ b/dozer-sql/src/builder/mod.rs @@ -83,6 +83,7 @@ pub fn statement_to_pipeline( udfs: Vec, runtime: Arc, ) -> Result { + dozer_sql_expression::wasm::validate_udfs(&udfs)?; let dialect = DozerDialect {}; let mut ctx = QueryContext::new(udfs, runtime); let is_top_select = true; diff --git a/dozer-sql/src/tests/mod.rs b/dozer-sql/src/tests/mod.rs index 0ed1750729..67937e06ce 100644 --- a/dozer-sql/src/tests/mod.rs +++ b/dozer-sql/src/tests/mod.rs @@ -1,2 +1,3 @@ mod builder_test; pub mod utils; +mod wasm_udf; diff --git a/dozer-sql/src/tests/wasm_udf.rs b/dozer-sql/src/tests/wasm_udf.rs new file mode 100644 index 0000000000..3bee89ef2e --- /dev/null +++ b/dozer-sql/src/tests/wasm_udf.rs @@ -0,0 +1,399 @@ +use std::{ + collections::{HashMap, VecDeque}, + sync::Arc, +}; + +use dozer_core::{ + app::{App, AppPipeline}, + appsource::{AppSourceManager, AppSourceMappings}, + channels::ProcessorChannelForwarder, + dag_schemas::{DagHaveSchemas, DagSchemas}, + daggy::{ + petgraph::visit::{EdgeRef, Topo}, + NodeIndex, Walker, + }, + event::EventHub, + node::{OutputPortDef, OutputPortType, PortHandle, Processor, Source, SourceFactory}, + NodeKind, DEFAULT_PORT_HANDLE, +}; +use dozer_types::{ + errors::internal::BoxedError, + models::{ + config::Config, + udf_config::{UdfConfig, UdfType}, + }, + types::{ + Field, FieldDefinition, FieldType, Operation, Record, Schema, SourceDefinition, + TableOperation, + }, +}; +use tokio::runtime::Runtime; + +use super::{builder_test::TestSinkFactory, utils::create_test_runtime}; +use crate::builder::statement_to_pipeline; + +#[derive(Debug)] +struct Input; +impl SourceFactory for Input { + fn get_output_schema(&self, _: &PortHandle) -> Result { + Ok(Schema { + fields: vec![ + FieldDefinition::new( + "field1".into(), + FieldType::Int, + true, + SourceDefinition::Table { + name: "input_rows".into(), + connection: "fixtures".into(), + }, + ), + FieldDefinition::new( + "field2".into(), + FieldType::Int, + false, + SourceDefinition::Table { + name: "input_rows".into(), + connection: "fixtures".into(), + }, + ), + ], + ..Default::default() + }) + } + fn get_output_port_name(&self, _: &PortHandle) -> String { + "input_rows".into() + } + fn get_output_ports(&self) -> Vec { + vec![OutputPortDef::new( + DEFAULT_PORT_HANDLE, + OutputPortType::Stateless, + )] + } + fn build( + &self, + _: HashMap, + _: EventHub, + _: Option>, + ) -> Result, BoxedError> { + unreachable!("test operations are supplied directly") + } +} +#[derive(Default)] +struct Output(Vec); +impl ProcessorChannelForwarder for Output { + fn send(&mut self, operation: TableOperation) { + self.0.push(operation); + } +} +type Endpoint = (NodeIndex, PortHandle); +struct Pipeline { + processors: HashMap>, + routes: HashMap>, + source: NodeIndex, + sink: NodeIndex, + schema: Schema, + _runtime: Arc, +} +impl Pipeline { + fn new(sql: &str, udfs: Vec) -> Result { + let runtime = create_test_runtime(); + let mut pipeline = AppPipeline::new(Default::default()); + let context = statement_to_pipeline( + sql, + &mut pipeline, + Some("result".into()), + udfs, + runtime.clone(), + )?; + let result = &context.output_tables_map["result"]; + pipeline.add_sink( + Box::new(TestSinkFactory::new(vec![DEFAULT_PORT_HANDLE])), + "capture".into(), + ); + pipeline.connect_nodes( + result.node.clone(), + result.port, + "capture".into(), + DEFAULT_PORT_HANDLE, + ); + let mut sources = AppSourceManager::new(); + sources.add( + Box::new(Input), + AppSourceMappings::new( + "fixtures".into(), + HashMap::from([("input_rows".into(), DEFAULT_PORT_HANDLE)]), + ), + )?; + let mut app = App::new(sources); + app.add_pipeline(pipeline); + let dag = runtime.block_on(DagSchemas::new(app.into_dag()?))?; + let mut processors = HashMap::new(); + let mut source = None; + let mut sink = None; + let mut schema = None; + for index in Topo::new(dag.graph()).iter(dag.graph()) { + match &dag.graph()[index].kind { + NodeKind::Source(_) => source = Some(index), + NodeKind::Processor(factory) => { + processors.insert( + index, + runtime.block_on(factory.build( + dag.get_node_input_schemas(index), + dag.get_node_output_schemas(index), + EventHub::new(1), + ))?, + ); + } + NodeKind::Sink(_) => { + sink = Some(index); + schema = Some(dag.get_node_input_schemas(index)[&DEFAULT_PORT_HANDLE].clone()); + } + } + } + let mut routes: HashMap> = HashMap::new(); + for edge in dag.graph().graph().edge_references() { + routes + .entry((edge.source(), edge.weight().output_port)) + .or_default() + .push((edge.target(), edge.weight().input_port)); + } + Ok(Self { + processors, + routes, + source: source.unwrap(), + sink: sink.unwrap(), + schema: schema.unwrap(), + _runtime: runtime, + }) + } + + fn feed(&mut self, operation: Operation) -> Result, BoxedError> { + let mut queue = VecDeque::from([( + self.source, + TableOperation::without_id(operation, DEFAULT_PORT_HANDLE), + )]); + let mut emitted = Vec::new(); + while let Some((from, operation)) = queue.pop_front() { + for &(to, port) in &self.routes[&(from, operation.port)] { + let operation = TableOperation { + port, + op: operation.op.clone(), + id: operation.id, + }; + if to == self.sink { + emitted.push(operation.op); + } else { + let mut output = Output::default(); + self.processors + .get_mut(&to) + .unwrap() + .process(operation, &mut output)?; + queue.extend(output.0.into_iter().map(|operation| (to, operation))); + } + } + } + Ok(emitted) + } +} + +fn example_config() -> Config { + let project = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("../examples/wasm-udf"); + let mut config: Config = dozer_types::serde_yaml::from_str( + &std::fs::read_to_string(project.join("dozer-config.yaml")).unwrap(), + ) + .unwrap(); + // Model launching Dozer from the example project directory without changing + // this test process's working directory (tests run concurrently). + for udf in &mut config.udfs { + if let UdfType::Wasm(wasm) = &mut udf.config { + wasm.path = project.join(&wasm.path).to_str().unwrap().into(); + } + } + config +} +fn row(a: i64, b: i64) -> Record { + Record::new(vec![Field::Int(a), Field::Int(b)]) +} +fn value(a: i64) -> Record { + Record::new(vec![Field::Int(a)]) +} + +#[test] +fn wasm_assemblyscript_streaming_demo() { + let config = example_config(); + let mut pipeline = Pipeline::new(config.sql.as_deref().unwrap(), config.udfs).unwrap(); + assert_eq!(pipeline.schema.fields[0].typ, FieldType::Int); + assert!(pipeline.schema.fields[0].nullable); + println!("AssemblyScript .wasm -> original YAML mapping -> SQL planner -> real DAG processors"); + let events = [ + ( + Operation::Insert { new: row(3, 4) }, + Operation::Insert { new: value(7) }, + ), + ( + Operation::Update { + old: row(3, 4), + new: row(10, -2), + }, + Operation::Update { + old: value(7), + new: value(8), + }, + ), + ( + Operation::Delete { old: row(10, -2) }, + Operation::Delete { old: value(8) }, + ), + ( + Operation::Insert { + new: Record::new(vec![Field::Null, Field::Int(1)]), + }, + Operation::Insert { + new: Record::new(vec![Field::Null]), + }, + ), + ]; + for (input, expected) in events { + let output = pipeline.feed(input.clone()).unwrap(); + println!("{input:?} -> {output:?}"); + assert_eq!(output, vec![expected]); + } +} + +#[test] +fn wasm_assemblyscript_selection_and_case_insensitive_sql_call() { + let mut config = example_config(); + config.udfs[0].name = "My_Function".into(); + if let UdfType::Wasm(wasm) = &mut config.udfs[0].config { + wasm.export = Some("my_function".into()); + } + let mut pipeline = Pipeline::new("SELECT MY_FUNCTION(field1, field2) AS total FROM input_rows WHERE my_function(field1, field2) > 5", config.udfs).unwrap(); + assert!(pipeline + .feed(Operation::Insert { new: row(1, 2) }) + .unwrap() + .is_empty()); + assert_eq!( + pipeline.feed(Operation::Insert { new: row(3, 4) }).unwrap(), + vec![Operation::Insert { new: value(7) }] + ); +} + +#[test] +fn wasm_bad_configuration_and_wrong_sql_types_fail_before_streaming() { + let mut config = example_config(); + if let UdfType::Wasm(wasm) = &mut config.udfs[0].config { + wasm.export = Some("missing".into()); + } + assert!(Pipeline::new("SELECT field1 FROM input_rows", config.udfs).is_err()); + assert!(Pipeline::new( + "SELECT my_function('not a pointer', field2) FROM input_rows", + example_config().udfs + ) + .is_err()); + assert!(Pipeline::new( + "SELECT my_function(field1) FROM input_rows", + example_config().udfs + ) + .is_err()); +} + +#[test] +fn wasm_assemblyscript_abort_does_not_poison_later_rows() { + let mut config = example_config(); + if let UdfType::Wasm(wasm) = &mut config.udfs[0].config { + wasm.export = Some("checked_divide".into()); + } + let mut pipeline = Pipeline::new( + "SELECT my_function(field1, field2) FROM input_rows", + config.udfs, + ) + .unwrap(); + assert!(pipeline + .feed(Operation::Insert { new: row(6, 0) }) + .unwrap_err() + .to_string() + .contains("AssemblyScript abort")); + assert_eq!( + pipeline.feed(Operation::Insert { new: row(6, 2) }).unwrap(), + vec![Operation::Insert { new: value(3) }] + ); +} + +#[test] +fn wasm_scalar_calls_reject_modifiers_wildcards_and_named_arguments() { + for call in [ + "my_function(DISTINCT field1, field2)", + "my_function(field1, field2) OVER ()", + "my_function(field1, field2 ORDER BY field1)", + "my_function(field1, field2)", + "my_function(*, field2)", + "my_function(x => field1, y => field2)", + ] { + let result = Pipeline::new( + &format!("SELECT {call} FROM input_rows"), + example_config().udfs, + ); + let Err(error) = result else { + panic!("accepted unsupported scalar call: {call}"); + }; + assert!( + error.to_string().contains("scalar call") + || error.to_string().contains("Unsupported SQL function arg"), + "{call}: {error}" + ); + } +} + +#[test] +fn wasm_computed_nullable_arguments_keep_nullable_output_schema() { + let mut pipeline = Pipeline::new( + "SELECT my_function(field1 + 1, field2) FROM input_rows", + example_config().udfs, + ) + .unwrap(); + assert!(pipeline.schema.fields[0].nullable); + assert_eq!( + pipeline + .feed(Operation::Insert { + new: Record::new(vec![Field::Null, Field::Int(1)]) + }) + .unwrap(), + vec![Operation::Insert { + new: Record::new(vec![Field::Null]) + }] + ); + assert_eq!( + pipeline.feed(Operation::Insert { new: row(2, 3) }).unwrap(), + vec![Operation::Insert { new: value(6) }] + ); +} + +#[test] +fn wasm_assemblyscript_mutable_globals_do_not_change_retractions() { + let mut config = example_config(); + if let UdfType::Wasm(wasm) = &mut config.udfs[0].config { + wasm.export = Some("isolated_counter".into()); + } + let mut pipeline = + Pipeline::new("SELECT my_function(field1) FROM input_rows", config.udfs).unwrap(); + assert_eq!( + pipeline.feed(Operation::Insert { new: row(4, 0) }).unwrap(), + vec![Operation::Insert { new: value(5) }] + ); + assert_eq!( + pipeline + .feed(Operation::Update { + old: row(4, 0), + new: row(5, 0) + }) + .unwrap(), + vec![Operation::Update { + old: value(5), + new: value(6) + }] + ); + assert_eq!( + pipeline.feed(Operation::Delete { old: row(5, 0) }).unwrap(), + vec![Operation::Delete { old: value(6) }] + ); +} diff --git a/dozer-types/src/models/config.rs b/dozer-types/src/models/config.rs index 3d8f981758..61aff4fae3 100644 --- a/dozer-types/src/models/config.rs +++ b/dozer-types/src/models/config.rs @@ -63,7 +63,12 @@ pub struct Config { pub telemetry: TelemetryConfig, /// UDF specific configuration (eg. !Onnx) - #[serde(default, skip_serializing_if = "Vec::is_empty")] + #[serde( + default, + skip_serializing_if = "Vec::is_empty", + deserialize_with = "super::udf_config::deserialize_udfs" + )] + #[schemars(with = "super::udf_config::UdfConfigurations")] pub udfs: Vec, /// Lambda functions. diff --git a/dozer-types/src/models/udf_config.rs b/dozer-types/src/models/udf_config.rs index 512fba3a57..3c62240900 100644 --- a/dozer-types/src/models/udf_config.rs +++ b/dozer-types/src/models/udf_config.rs @@ -16,6 +16,7 @@ pub struct UdfConfig { pub enum UdfType { Onnx(OnnxConfig), JavaScript(JavaScriptConfig), + Wasm(WasmConfig), } #[derive(Debug, Serialize, Deserialize, JsonSchema, Eq, PartialEq, Clone)] @@ -31,3 +32,131 @@ pub struct JavaScriptConfig { /// path to the module file pub module: String, } + +/// Numeric WebAssembly scalar function. Relative paths use the Dozer project directory. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct WasmConfig { + pub path: String, + /// Export name; defaults to the configured SQL function name. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub export: Option, + /// Instruction budget including module initialization, for each SQL evaluation. + #[serde(default = "default_wasm_fuel")] + #[schemars(range(min = 1, max = 10000000))] + pub fuel: u64, + /// Maximum linear memory size, in bytes. + #[serde(default = "default_wasm_memory")] + #[schemars(range(min = 1, max = 67108864))] + pub memory_limit: usize, +} + +pub fn default_wasm_fuel() -> u64 { + 1_000_000 +} + +pub fn default_wasm_memory() -> usize { + 16 * 1024 * 1024 +} + +#[derive(JsonSchema)] +#[serde(untagged)] +pub enum UdfConfigurations { + List(Vec), + WasmMap(WasmUdfMap), +} + +#[derive(JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct WasmUdfMap { + pub wasm: std::collections::BTreeMap, +} + +impl WasmConfig { + pub fn new(path: String) -> Self { + Self { + path, + export: None, + fuel: default_wasm_fuel(), + memory_limit: default_wasm_memory(), + } + } +} + +/// Accept the original `udfs: { wasm: { function: path } }` proposal as well as +/// the existing list of tagged UDF configurations. A visitor preserves YAML tags +/// in list entries, including existing !Onnx and !JavaScript configurations. +pub fn deserialize_udfs<'de, D>(deserializer: D) -> Result, D::Error> +where + D: crate::serde::Deserializer<'de>, +{ + use crate::serde::de::{value::SeqAccessDeserializer, Error, MapAccess, SeqAccess, Visitor}; + use std::{collections::BTreeMap, fmt}; + + struct WasmPaths(BTreeMap); + impl<'de> Deserialize<'de> for WasmPaths { + fn deserialize>( + deserializer: D, + ) -> Result { + struct PathsVisitor; + impl<'de> Visitor<'de> for PathsVisitor { + type Value = WasmPaths; + + fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result { + formatter.write_str("a wasm function-to-path mapping") + } + + fn visit_map>(self, mut map: A) -> Result { + let mut paths = BTreeMap::new(); + while let Some((name, path)) = map.next_entry::()? { + if paths.insert(name.clone(), path).is_some() { + return Err(A::Error::custom(format!( + "duplicate wasm function `{name}`" + ))); + } + } + Ok(WasmPaths(paths)) + } + } + deserializer.deserialize_map(PathsVisitor) + } + } + + struct UdfsVisitor; + impl<'de> Visitor<'de> for UdfsVisitor { + type Value = Vec; + + fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result { + formatter.write_str("a UDF list or a wasm function-to-path mapping") + } + + fn visit_seq>(self, seq: A) -> Result { + Vec::deserialize(SeqAccessDeserializer::new(seq)) + } + + fn visit_map>(self, mut map: A) -> Result { + let mut result = Vec::new(); + let mut seen = false; + while let Some(key) = map.next_key::()? { + if key != "wasm" { + return Err(A::Error::unknown_field(&key, &["wasm"])); + } + if seen { + return Err(A::Error::duplicate_field("wasm")); + } + seen = true; + for (name, path) in map.next_value::()?.0 { + result.push(UdfConfig { + name, + config: UdfType::Wasm(WasmConfig::new(path)), + }); + } + } + if !seen { + return Err(A::Error::missing_field("wasm")); + } + Ok(result) + } + } + deserializer.deserialize_any(UdfsVisitor) +} diff --git a/dozer-types/src/tests/udf_yaml_deserialize.rs b/dozer-types/src/tests/udf_yaml_deserialize.rs index 6cec087a5b..d866df9072 100644 --- a/dozer-types/src/tests/udf_yaml_deserialize.rs +++ b/dozer-types/src/tests/udf_yaml_deserialize.rs @@ -1,4 +1,59 @@ use crate::models::udf_config::{OnnxConfig, UdfConfig, UdfType}; +use crate::models::{config::Config, udf_config::WasmConfig}; + +#[test] +fn wasm_original_mapping_and_tagged_list() { + let original: Config = serde_yaml::from_str( + "version: 1\napp_name: wasm\nudfs:\n wasm:\n my_function: udfs/wasm/my_function\n", + ) + .unwrap(); + assert_eq!( + original.udfs, + vec![UdfConfig { + name: "my_function".into(), + config: UdfType::Wasm(WasmConfig::new("udfs/wasm/my_function".into())) + }] + ); + let tagged: Config = serde_yaml::from_str("version: 1\napp_name: wasm\nudfs:\n - name: my_function\n config: !Wasm\n path: udfs/wasm/my_function\n").unwrap(); + assert_eq!(original, tagged); + assert_eq!( + original, + serde_yaml::from_str(&serde_yaml::to_string(&original).unwrap()).unwrap() + ); +} + +#[test] +fn wasm_mapping_does_not_hide_invalid_fields_or_break_existing_tags() { + assert!(serde_yaml::from_str::( + "version: 1\napp_name: wasm\nudfs:\n wasmm:\n f: a.wasm" + ) + .is_err()); + let config: Config = serde_yaml::from_str("version: 1\napp_name: wasm\nudfs:\n - name: model\n config: !Onnx\n path: model.onnx\n - name: js\n config: !JavaScript\n module: file.js\n").unwrap(); + assert!(matches!(config.udfs[0].config, UdfType::Onnx(_))); + assert!(matches!(config.udfs[1].config, UdfType::JavaScript(_))); +} + +#[test] +fn wasm_mapping_rejects_duplicate_functions_and_requires_wasm_key() { + let duplicate = serde_yaml::from_str::( + "version: 1\napp_name: wasm\nudfs:\n wasm:\n f: first.wasm\n f: second.wasm\n", + ) + .unwrap_err(); + assert!(duplicate + .to_string() + .contains("duplicate wasm function `f`")); + assert!( + serde_yaml::from_str::("version: 1\napp_name: wasm\nudfs: {}") + .unwrap_err() + .to_string() + .contains("missing field `wasm`") + ); + for empty in ["[]", "{wasm: {}}"] { + let config: Config = + serde_yaml::from_str(&format!("version: 1\napp_name: wasm\nudfs: {empty}")).unwrap(); + assert!(config.udfs.is_empty()); + } +} #[test] fn standard() { diff --git a/examples/wasm-udf/.gitignore b/examples/wasm-udf/.gitignore new file mode 100644 index 0000000000..c2658d7d1b --- /dev/null +++ b/examples/wasm-udf/.gitignore @@ -0,0 +1 @@ +node_modules/ diff --git a/examples/wasm-udf/README.md b/examples/wasm-udf/README.md new file mode 100644 index 0000000000..3628a0ea9d --- /dev/null +++ b/examples/wasm-udf/README.md @@ -0,0 +1,120 @@ +# WebAssembly scalar UDFs + +Compile an AssemblyScript function, register the `.wasm` file, then call it from +SQL. Dozer executes the binary with wasmi; Node and JavaScript bindings are not +part of the execution path. + +## Build and configure + +From this example directory: + +```sh +npm ci --ignore-scripts +npm run build +``` + +The pinned AssemblyScript compiler produces `udfs/wasm/my_function.wasm` from +`assembly/index.ts`. The binary and readable `.wat` are included so Rust tests +do not need Node or download a compiler. + +Add this to the project's `dozer-config.yaml`: + +```yaml +udfs: + wasm: + my_function: udfs/wasm/my_function +``` + +Run Dozer from the project directory containing `udfs/`. Paths use Dozer's +existing **working directory** convention, including when `--config` points +elsewhere; they are not relative to each configuration file. Absolute paths are +also accepted. A path without an extension receives `.wasm` automatically. +Configure a source named `input_rows` with integer fields `field1` and `field2`, +then use: + +```sql +SELECT my_function(field1, field2) INTO result FROM input_rows; +``` + +This example's configuration does not invent a live connector. The integration +test supplies synthetic source operations to the actual SQL planner and processor +factories. + +The existing list format also works, and supports an export alias and limits: + +```yaml +udfs: + - name: scaled + config: !Wasm + path: udfs/wasm/my_function.wasm + export: scale + fuel: 1000000 + memory_limit: 16777216 +``` + +SQL names are case-insensitive; WASM export names are case-sensitive and default +to the configured name. Export one fixed-arity function for each registered UDF. +Calls take positional scalar expressions; named arguments, wildcards, window or +aggregate modifiers, and SQL return-type overrides are rejected. +Do not compile with `--exportStart`: use the compiler's normal implicit module +initialization. AssemblyScript optional parameters and object bindings are not +part of this ABI. + +## Values and isolation + +| WASM type | SQL input | SQL result | +| --- | --- | --- | +| `i32` | Integer fields within the signed 32-bit range | `Int` | +| `i64` | Integer fields within the signed 64-bit range | `Int` | +| `f32` | `Float`, rounded to finite single precision | `Float` | +| `f64` | Finite `Float` | `Float` | + +Use an explicit SQL `CAST(... AS FLOAT)` when passing an integer expression to a +floating-point parameter. Unsigned and 128-bit integer fields are checked before +conversion. NaN, infinities and out-of-range arguments are errors. A NULL argument +returns SQL NULL without invoking the function; output nullability follows the +arguments (computed expressions are conservatively nullable). Each function must +return exactly one numeric value. Strings, objects, +pointers, references and multiple return values are unsupported. + +WASM arithmetic keeps its standard semantics: integer addition can wrap inside +the guest. Host argument conversions never wrap; use AssemblyScript checks or +assertions when arithmetic overflow should fail. Division traps and assertions +are reported as SQL expression errors. + +Modules are validated during pipeline setup, including exports, parameter counts, +types, imports and initialization. A compiled module is retained by each SQL +expression, while every evaluation gets fresh memory, globals, fuel and limits. +An update/delete therefore recomputes the same function as the original insert; +mutable guest state cannot accumulate across records. The only allowed import is +AssemblyScript's `env.abort(i32, i32, i32, i32)`, which raises an error without +reading guest pointers. Filesystem, network, clocks, random sources and WASI imports +are unavailable. + +The default per-evaluation budget is 1,000,000 fuel units and 16 MiB of linear +memory, including initialization. Configurations may use at most 10,000,000 fuel +and 64 MiB. Modules are at most 10 MiB; one memory, one table and 4,096 table +elements are allowed. Failed memory/table growth traps. Fuel exhaustion and guest +traps return errors; a later record receives a new instance and can still run. +Fuel bounds execution, not compilation time. Fresh instantiation prioritizes +repeatable streaming results over the throughput of a stateful instance pool. + +## Reproduce the local verification + +From the repository root, with its normal Rust/protoc environment: + +```sh +cargo test --locked -p dozer-sql-expression --no-default-features --lib wasm::tests +cargo test --locked -p dozer-sql --no-default-features --lib tests::wasm_udf +cargo test --locked -p dozer-sql --no-default-features --lib \ + tests::wasm_udf::wasm_assemblyscript_streaming_demo -- --exact --nocapture +``` + +The demo uses the checked-in AssemblyScript binary, original mapping configuration, +real SQL schema propagation and processor factories. It checks insert, update, +delete and NULL events. A synchronous queue replaces connector transport; it does +not demonstrate a live connector or checkpoint recovery. + +References: [AssemblyScript compiler](https://www.assemblyscript.org/compiler.html), +[AssemblyScript imports](https://www.assemblyscript.org/concepts.html#special-imports), +[wasmi 0.31.2](https://docs.rs/wasmi/0.31.2/wasmi/). diff --git a/examples/wasm-udf/assembly/index.ts b/examples/wasm-udf/assembly/index.ts new file mode 100644 index 0000000000..8d8662ead9 --- /dev/null +++ b/examples/wasm-udf/assembly/index.ts @@ -0,0 +1,18 @@ +export function my_function(field1: i64, field2: i64): i64 { + return field1 + field2; +} + +export function scale(value: f64, factor: f64): f64 { + return value * factor; +} + +export function checked_divide(value: i32, divisor: i32): i32 { + assert(divisor != 0, "division by zero"); + return value / divisor; +} + +let counter: i64 = 0; +export function isolated_counter(value: i64): i64 { + counter += 1; + return value + counter; +} diff --git a/examples/wasm-udf/dozer-config.yaml b/examples/wasm-udf/dozer-config.yaml new file mode 100644 index 0000000000..3acaee36c3 --- /dev/null +++ b/examples/wasm-udf/dozer-config.yaml @@ -0,0 +1,6 @@ +version: 1 +app_name: wasm-udf-example +udfs: + wasm: + my_function: udfs/wasm/my_function +sql: SELECT my_function(field1, field2) INTO result FROM input_rows; diff --git a/examples/wasm-udf/package-lock.json b/examples/wasm-udf/package-lock.json new file mode 100644 index 0000000000..e8eb322195 --- /dev/null +++ b/examples/wasm-udf/package-lock.json @@ -0,0 +1,54 @@ +{ + "name": "dozer-wasm-udf-example", + "lockfileVersion": 3, + "requires": true, + "packages": { + "": { + "name": "dozer-wasm-udf-example", + "devDependencies": { + "assemblyscript": "0.27.31" + } + }, + "node_modules/assemblyscript": { + "version": "0.27.31", + "resolved": "https://registry.npmjs.org/assemblyscript/-/assemblyscript-0.27.31.tgz", + "integrity": "sha512-Ra8kiGhgJQGZcBxjtMcyVRxOEJZX64kd+XGpjWzjcjgxWJVv+CAQO0aDBk4GQVhjYbOkATarC83mHjAVGtwPBQ==", + "dev": true, + "license": "Apache-2.0", + "dependencies": { + "binaryen": "116.0.0-nightly.20240114", + "long": "^5.2.1" + }, + "bin": { + "asc": "bin/asc.js", + "asinit": "bin/asinit.js" + }, + "engines": { + "node": ">=16", + "npm": ">=7" + }, + "funding": { + "type": "opencollective", + "url": "https://opencollective.com/assemblyscript" + } + }, + "node_modules/binaryen": { + "version": "116.0.0-nightly.20240114", + "resolved": "https://registry.npmjs.org/binaryen/-/binaryen-116.0.0-nightly.20240114.tgz", + "integrity": "sha512-0GZrojJnuhoe+hiwji7QFaL3tBlJoA+KFUN7ouYSDGZLSo9CKM8swQX8n/UcbR0d1VuZKU+nhogNzv423JEu5A==", + "dev": true, + "license": "Apache-2.0", + "bin": { + "wasm-opt": "bin/wasm-opt", + "wasm2js": "bin/wasm2js" + } + }, + "node_modules/long": { + "version": "5.3.2", + "resolved": "https://registry.npmjs.org/long/-/long-5.3.2.tgz", + "integrity": "sha512-mNAgZ1GmyNhD7AuqnTG3/VQ26o760+ZYBPKjPvugO8+nLbYfX6TVpJPseBvopbdY+qpZ/lKUnmEc1LeZYS3QAA==", + "dev": true, + "license": "Apache-2.0" + } + } +} diff --git a/examples/wasm-udf/package.json b/examples/wasm-udf/package.json new file mode 100644 index 0000000000..266b2ca883 --- /dev/null +++ b/examples/wasm-udf/package.json @@ -0,0 +1,10 @@ +{ + "name": "dozer-wasm-udf-example", + "private": true, + "scripts": { + "build": "asc assembly/index.ts --runtime stub --outFile udfs/wasm/my_function.wasm --textFile udfs/wasm/my_function.wat --optimize" + }, + "devDependencies": { + "assemblyscript": "0.27.31" + } +} diff --git a/examples/wasm-udf/udfs/wasm/my_function.wasm b/examples/wasm-udf/udfs/wasm/my_function.wasm new file mode 100644 index 0000000000000000000000000000000000000000..17aadc0ad681a6197178bb09c746d0a1071c1848 GIT binary patch literal 319 zcmXAkK~BRk5JmquP8!uvbvM>mirE99JV<&G8wlghcRGYeK z2Z1w^^9MOYf`q&})o*pd-9|p?uUP-Zdai>dppHnb7Fc0T{+bgjc%}9d`Xg5K2>ltW z8^PBltO6x^mSZ)r?Cmh{{|