Repository: ArroyoSystems/arroyo (language: rust)

Issue:
Sync UDFs: Unexpected Requirement for Parameter
### UDF Definition
```rust
/*
[dependencies]
chrono = "0.4"
*/
 
use arroyo_udf_plugin::udf;
use chrono::{Utc, Duration};
 
#[udf]
fn date_id_with_delta(delta: i64) -> String {
    let now = Utc::now();
    let result_date = now + Duration::days(delta);
    result_date.format("%Y%m%d").to_string()
}
```
### Query (table is a Kafka source)
```sql
SELECT a, b, date_id_with_delta(-1) AS date_str
FROM table
```
An error is thrown after the first row is processed:
 
```
ERROR arroyo_server_common: panicked at crates/arroyo-worker/src/arrow/mod.rs:67:31:
should be able to compute batch: ArrowError(InvalidArgumentError("all columns in a record batch must have the same length"), None) panic.file="crates/arroyo-worker/src/arrow/mod.rs" panic.line=67 panic.column=31
```

### When changing the UDF to include an extra parameter that isn’t used:
```rust
/*
[dependencies]
chrono = "0.4"
*/
 
use arroyo_udf_plugin::udf;
use chrono::{Utc, Duration};
 
#[udf]
fn date_id_with_delta(x: f64, delta: i64) -> String {
    let now = Utc::now();
    let result_date = now + Duration::days(delta);
    result_date.format("%Y%m%d").to_string()
}
```
 
### And adjust the query to:
 
```sql
SELECT a, b, date_id_with_delta(x, -1) AS date_str
FROM table
```
 
The query runs without any issues.



Discussion / hints:

Fix patch:
diff --git a/crates/arroyo-udf/arroyo-udf-host/src/lib.rs b/crates/arroyo-udf/arroyo-udf-host/src/lib.rs
index cc600b829..25bcbe2cb 100644
--- a/crates/arroyo-udf/arroyo-udf-host/src/lib.rs
+++ b/crates/arroyo-udf/arroyo-udf-host/src/lib.rs
@@ -390,6 +390,8 @@ impl ScalarUDFImpl for SyncUdfDylib {
             .max()
             .unwrap();
 
+        let all_scalars = args.iter().all(|c| matches!(c, ColumnarValue::Scalar(_)));
+
         let args = args
             .iter()
             .map(|arg| arg.clone().into_array(num_rows).unwrap().to_data())
@@ -402,7 +404,15 @@ impl ScalarUDFImpl for SyncUdfDylib {
         match result {
             RunResult::Ok(FfiArraySchema(array, schema)) => {
                 let result_array = unsafe { from_ffi(array, &schema).unwrap() };
-                Ok(ColumnarValue::Array(make_array(result_array)))
+
+                let array = make_array(result_array);
+                if all_scalars {
+                    Ok(ColumnarValue::Scalar(ScalarValue::try_from_array(
+                        &array, 0,
+                    )?))
+                } else {
+                    Ok(ColumnarValue::Array(array))
+                }
             }
             RunResult::Err => {
                 panic!("panic in UDF {}", self.name);


Test patch: