Repository: ArroyoSystems/arroyo (language: rust)

Issue:
Compilation error when applying non-optional UDF to optional inputs
This query produces a compilation error:

```rust
// udf
fn hello(i: i64) -> i64 {
    i
}
```

```sql
-- query
select hello(person.id) 
from nexmark;
```

this produces the following rust code:

```rust
pub struct generated_struct_9525578188421819516 {
    pub hello_nexmark100_person_id__: i64,
}

generated_struct_9525578188421819516 {
    hello_nexmark100_person_id__: {
        (|| {
            let __0 = (arg.person.clone().map(|arg| arg.id.clone()))?;
            udfs::hello(__0)
        })()
    },
},
```

which fails to compile with

```
error[E0277]: the `?` operator can only be used in a closure that returns `Result` or `Option` (or another type that implements `FromResidual`)
   --> pipeline/src/main.rs:123:109
    |
122 | ...  (|| {
    |       -- this function should return `Result` or `Option` to accept `?`
123 | ...      let __0 = (arg.person.clone().map(|arg| arg.id.clone()))?;
    |                                                                  ^ cannot use the `?` operator in a closure that returns `i64`
    |
    = help: the trait `FromResidual<std::option::Option<Infallible>>` is not implemented for `i64`
```

The expected behavior here is given a nullable input (which `person.id` should be, as it's a traversal through a nullable field), a UDF that has no optional inputs or return types should be called only on non-null inputs, and should produce optional outputs.

Instead, datafusion is planning it as having a non-null return type, and we're not properly wrapping the result of the UDF call in a `Some`.


Discussion / hints:

Fix patch:
diff --git a/arroyo-sql/src/expressions.rs b/arroyo-sql/src/expressions.rs
index fe364eadd..267c62f99 100644
--- a/arroyo-sql/src/expressions.rs
+++ b/arroyo-sql/src/expressions.rs
@@ -2458,7 +2458,8 @@ impl RustUdfExpression {
                 || self
                     .args
                     .iter()
-                    .any(|(t, e)| t.is_optional() && !e.nullable()),
+                    // nullable if there are non-null UDF params with nullable arguments
+                    .any(|(t, e)| !t.is_optional() && e.nullable()),
         )
     }
 }
diff --git a/arroyo-sql/src/json_schema.rs b/arroyo-sql/src/json_schema.rs
index 7cccd8ebc..1fc2fe725 100644
--- a/arroyo-sql/src/json_schema.rs
+++ b/arroyo-sql/src/json_schema.rs
@@ -7,7 +7,7 @@ use arrow_schema::DataType;
 use quote::{format_ident, quote};
 use schemars::schema::{RootSchema, Schema};
 use syn::{parse_str, Type};
-use tracing::log::warn;
+use tracing::warn;
 use typify::{TypeDetails, TypeSpace, TypeSpaceSettings};
 
 use crate::types::{StructDef, StructField, TypeDef};
diff --git a/build_dir/Cargo.lock b/build_dir/Cargo.lock
index e93818cf3..accf4c6f8 100644
--- a/build_dir/Cargo.lock
+++ b/build_dir/Cargo.lock
@@ -308,7 +308,7 @@ dependencies = [
 
 [[package]]
 name = "arroyo-macro"
-version = "0.3.0"
+version = "0.4.0"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -317,7 +317,7 @@ dependencies = [
 
 [[package]]
 name = "arroyo-metrics"
-version = "0.3.0"
+version = "0.4.0"
 dependencies = [
  "arroyo-types",
  "prometheus",
@@ -325,7 +325,7 @@ dependencies = [
 
 [[package]]
 name = "arroyo-rpc"
-version = "0.3.0"
+version = "0.4.0"
 dependencies = [
  "arroyo-types",
  "bincode 2.0.0-rc.3",
@@ -339,7 +339,7 @@ dependencies = [
 
 [[package]]
 name = "arroyo-server-common"
-version = "0.3.0"
+version = "0.4.0"
 dependencies = [
  "arroyo-types",
  "axum",
@@ -366,7 +366,7 @@ dependencies = [
 
 [[package]]
 name = "arroyo-state"
-version = "0.3.0"
+version = "0.4.0"
 dependencies = [
  "anyhow",
  "arrow",
@@ -395,7 +395,7 @@ dependencies = [
 
 [[package]]
 name = "arroyo-types"
-version = "0.3.0"
+version = "0.4.0"
 dependencies = [
  "anyhow",
  "base64 0.21.0",
@@ -406,7 +406,7 @@ dependencies = [
 
 [[package]]
 name = "arroyo-worker"
-version = "0.3.0"
+version = "0.4.0"
 dependencies = [
  "anyhow",
  "arroyo-macro",
diff --git a/build_dir/wasm-fns/Cargo.lock b/build_dir/wasm-fns/Cargo.lock
index fe1ed8fd1..1fd286c4c 100644
--- a/build_dir/wasm-fns/Cargo.lock
+++ b/build_dir/wasm-fns/Cargo.lock
@@ -28,7 +28,7 @@ dependencies = [
 
 [[package]]
 name = "arroyo-types"
-version = "0.3.0"
+version = "0.4.0"
 dependencies = [
  "anyhow",
  "base64",


Test patch:
diff --git a/arroyo-sql/src/test.rs b/arroyo-sql/src/test.rs
index 581a7e1a7..18539a344 100644
--- a/arroyo-sql/src/test.rs
+++ b/arroyo-sql/src/test.rs
@@ -1,9 +1,10 @@
+use arrow_schema::DataType;
 use arroyo_connectors::{
     nexmark::{NexmarkConnector, NexmarkTable},
     Connector, EmptyConfig,
 };
 
-use crate::{parse_and_get_program, ArroyoSchemaProvider, SqlConfig};
+use crate::{parse_and_get_program, types::TypeDef, ArroyoSchemaProvider, SqlConfig};
 
 #[tokio::test]
 async fn test_parse() {
@@ -162,6 +163,9 @@ async fn test_udf() {
         .add_rust_udf("fn my_sqr(x: i64) -> i64 { x * x }")
         .unwrap();
 
+    let def = schema_provider.udf_defs.get("my_sqr").unwrap();
+    assert_eq!(def.ret, TypeDef::DataType(DataType::Int64, false));
+
     let sql = "SELECT my_sqr(bid.auction) FROM nexmark";
     parse_and_get_program(sql, schema_provider, SqlConfig::default())
         .await