Fix Type Coercion for UDF Arguments

datafusion/expr-common/src/signature.rs

@@ -498,7 +498,7 @@ impl Signature {
         }
     }
 
-    /// A specified number of numeric arguments
+    /// A specified number of string arguments
     pub fn string(arg_count: usize, volatility: Volatility) -> Self {
         Self {
             type_signature: TypeSignature::String(arg_count),

datafusion/expr/src/type_coercion/functions.rs

@@ -584,23 +584,21 @@ fn get_valid_types(
                 match target_type_class {
                     TypeSignatureClass::Native(native_type) => {
                         let target_type = native_type.native();
-                        if &logical_type == target_type {
-                            return target_type.default_cast_for(current_type);
-                        }
-
-                        if logical_type == NativeType::Null {
-                            return target_type.default_cast_for(current_type);
-                        }
-
-                        if target_type.is_integer() && logical_type.is_integer() {
-                            return target_type.default_cast_for(current_type);
-                        }
-
-                        internal_err!(
-                            "Expect {} but received {}",
-                            target_type_class,
-                            current_type
-                        )
+                        target_type.default_cast_for(current_type)
+                        // if &logical_type == target_type {
+                        //     return target_type.default_cast_for(current_type);
+                        // }
+                        // if logical_type == NativeType::Null {
+                        //     return target_type.default_cast_for(current_type);
+                        // }
+                        // if target_type.is_integer() && logical_type.is_integer() {
+                        //     return target_type.default_cast_for(current_type);
+                        // }
+                        // internal_err!(
+                        //     "Expect {} but received {}",
+                        //     target_type_class,
+                        //     current_type
+                        // )
                     }
                     // Not consistent with Postgres and DuckDB but to avoid regression we implicit cast string to timestamp
                     TypeSignatureClass::Timestamp

datafusion/functions/Cargo.toml

@@ -91,6 +91,7 @@ uuid = { version = "1.7", features = ["v4"], optional = true }
 [dev-dependencies]
 arrow = { workspace = true, features = ["test_utils"] }
 criterion = "0.5"
+datafusion = { workspace = true, default-features = false }
 rand = { workspace = true }
 tokio = { workspace = true, features = ["macros", "rt", "sync"] }
 

datafusion/functions/src/string/ascii.rs

@@ -19,9 +19,12 @@ use crate::utils::make_scalar_function;
 use arrow::array::{ArrayAccessor, ArrayIter, ArrayRef, AsArray, Int32Array};
 use arrow::datatypes::DataType;
 use arrow::error::ArrowError;
+use datafusion_common::types::logical_string;
 use datafusion_common::{internal_err, Result};
-use datafusion_expr::{ColumnarValue, Documentation};
-use datafusion_expr::{ScalarUDFImpl, Signature, Volatility};
+use datafusion_expr::{
+    ColumnarValue, Documentation, ScalarUDFImpl, Signature, TypeSignature,
+    TypeSignatureClass, Volatility,
+};
 use datafusion_macros::user_doc;
 use std::any::Any;
 use std::sync::Arc;
@@ -61,7 +64,15 @@ impl Default for AsciiFunc {
 impl AsciiFunc {
     pub fn new() -> Self {
         Self {
-            signature: Signature::string(1, Volatility::Immutable),
+            signature: Signature::one_of(
+                vec![
+                    TypeSignature::String(1),
+                    TypeSignature::Coercible(vec![TypeSignatureClass::Native(
+                        logical_string(),
+                    )]),
+                ],
+                Volatility::Immutable,
+            ),
         }
     }
 }
@@ -130,20 +141,23 @@ pub fn ascii(args: &[ArrayRef]) -> Result<ArrayRef> {
             let string_array = args[0].as_string_view();
             Ok(calculate_ascii(string_array)?)
         }
-        _ => internal_err!("Unsupported data type"),
+        other => internal_err!("Unsupported data type for ascii: {:?}", other),
     }
 }
 
 #[cfg(test)]
 mod tests {
+    use crate::expr_fn::ascii;
     use crate::string::ascii::AsciiFunc;
     use crate::utils::test::test_function;
-    use arrow::array::{Array, Int32Array};
+    use arrow::array::{Array, ArrayRef, Int32Array, RecordBatch, StringArray};
     use arrow::datatypes::DataType::Int32;
-    use datafusion_common::{Result, ScalarValue};
-    use datafusion_expr::{ColumnarValue, ScalarUDFImpl};
+    use datafusion::prelude::SessionContext;
+    use datafusion_common::{DFSchema, Result, ScalarValue};
+    use datafusion_expr::{col, lit, ColumnarValue, ScalarUDFImpl};
+    use std::sync::Arc;
 
-    macro_rules! test_ascii {
+    macro_rules! test_ascii_invoke {
         ($INPUT:expr, $EXPECTED:expr) => {
             test_function!(
                 AsciiFunc::new(),
@@ -174,12 +188,64 @@ mod tests {
         };
     }
 
+    fn ascii_array(input: ArrayRef) -> Result<ArrayRef> {
+        let batch = RecordBatch::try_from_iter([("c0", input)])?;
+        let df_schema = DFSchema::try_from(batch.schema())?;
+        let expr = ascii(col("c0"));
+        let physical_expr =
+            SessionContext::new().create_physical_expr(expr, &df_schema)?;
+        let result = match physical_expr.evaluate(&batch)? {
+            ColumnarValue::Array(result) => Ok(result),
+            _ => datafusion_common::internal_err!("ascii"),
+        }?;
+        Ok(result)
+    }
+
+    fn ascii_scalar(input: ScalarValue) -> Result<ScalarValue> {
+        let df_schema = DFSchema::empty();
+        let expr = ascii(lit(input));
+        let physical_expr =
+            SessionContext::new().create_physical_expr(expr, &df_schema)?;
+        let result = match physical_expr
+            .evaluate(&RecordBatch::new_empty(Arc::clone(df_schema.inner())))?
+        {
+            ColumnarValue::Scalar(result) => Ok(result),
+            _ => datafusion_common::internal_err!("ascii"),
+        }?;
+        Ok(result)
+    }
+
     #[test]
-    fn test_functions() -> Result<()> {
-        test_ascii!(Some(String::from("x")), Ok(Some(120)));
-        test_ascii!(Some(String::from("a")), Ok(Some(97)));
-        test_ascii!(Some(String::from("")), Ok(Some(0)));
-        test_ascii!(None, Ok(None));
+    fn test_ascii_invoke() -> Result<()> {
+        test_ascii_invoke!(Some(String::from("x")), Ok(Some(120)));
+        test_ascii_invoke!(Some(String::from("a")), Ok(Some(97)));
+        test_ascii_invoke!(Some(String::from("")), Ok(Some(0)));
+        test_ascii_invoke!(None, Ok(None));
+        Ok(())
+    }
+
+    #[test]
+    fn test_ascii_expr() -> Result<()> {
+        let input = Arc::new(StringArray::from(vec![Some("x")])) as ArrayRef;
+        let expected = Arc::new(Int32Array::from(vec![Some(120)])) as ArrayRef;
+        let result = ascii_array(input)?;
+        assert_eq!(&expected, &result);
+
+        let input = ScalarValue::Utf8(Some(String::from("x")));
+        let expected = ScalarValue::Int32(Some(120));
+        let result = ascii_scalar(input)?;
+        assert_eq!(&expected, &result);
+
+        let input = Arc::new(Int32Array::from(vec![Some(2)])) as ArrayRef;
+        let expected = Arc::new(Int32Array::from(vec![Some(50)])) as ArrayRef;
+        let result = ascii_array(input)?;
+        assert_eq!(&expected, &result);
+
+        let input = ScalarValue::Int32(Some(2));
+        let expected = ScalarValue::Int32(Some(50));
+        let result = ascii_scalar(input)?;
+        assert_eq!(&expected, &result);
+
         Ok(())
     }
 }

datafusion/functions/src/string/bit_length.rs

@@ -20,9 +20,12 @@ use arrow::datatypes::DataType;
 use std::any::Any;
 
 use crate::utils::utf8_to_int_type;
+use datafusion_common::types::logical_string;
 use datafusion_common::{exec_err, Result, ScalarValue};
-use datafusion_expr::{ColumnarValue, Documentation, Volatility};
-use datafusion_expr::{ScalarUDFImpl, Signature};
+use datafusion_expr::{
+    ColumnarValue, Documentation, ScalarUDFImpl, Signature, TypeSignature,
+    TypeSignatureClass, Volatility,
+};
 use datafusion_macros::user_doc;
 
 #[user_doc(
@@ -55,7 +58,15 @@ impl Default for BitLengthFunc {
 impl BitLengthFunc {
     pub fn new() -> Self {
         Self {
-            signature: Signature::string(1, Volatility::Immutable),
+            signature: Signature::one_of(
+                vec![
+                    TypeSignature::String(1),
+                    TypeSignature::Coercible(vec![TypeSignatureClass::Native(
+                        logical_string(),
+                    )]),
+                ],
+                Volatility::Immutable,
+            ),
         }
     }
 }

datafusion/functions/src/string/contains.rs

@@ -21,10 +21,12 @@ use arrow::compute::contains as arrow_contains;
 use arrow::datatypes::DataType;
 use arrow::datatypes::DataType::{Boolean, LargeUtf8, Utf8, Utf8View};
 use datafusion_common::exec_err;
+use datafusion_common::types::logical_string;
 use datafusion_common::DataFusionError;
 use datafusion_common::Result;
 use datafusion_expr::{
-    ColumnarValue, Documentation, ScalarUDFImpl, Signature, Volatility,
+    ColumnarValue, Documentation, ScalarUDFImpl, Signature, TypeSignature,
+    TypeSignatureClass, Volatility,
 };
 use datafusion_macros::user_doc;
 use std::any::Any;
@@ -59,7 +61,16 @@ impl Default for ContainsFunc {
 impl ContainsFunc {
     pub fn new() -> Self {
         Self {
-            signature: Signature::string(2, Volatility::Immutable),
+            signature: Signature::one_of(
+                vec![
+                    TypeSignature::String(2),
+                    TypeSignature::Coercible(vec![
+                        TypeSignatureClass::Native(logical_string()),
+                        TypeSignatureClass::Native(logical_string()),
+                    ]),
+                ],
+                Volatility::Immutable,
+            ),
         }
     }
 }

datafusion/functions/src/string/ends_with.rs

@@ -22,9 +22,12 @@ use arrow::array::ArrayRef;
 use arrow::datatypes::DataType;
 
 use crate::utils::make_scalar_function;
+use datafusion_common::types::logical_string;
 use datafusion_common::{internal_err, Result};
-use datafusion_expr::{ColumnarValue, Documentation, Volatility};
-use datafusion_expr::{ScalarUDFImpl, Signature};
+use datafusion_expr::{
+    ColumnarValue, Documentation, ScalarUDFImpl, Signature, TypeSignature,
+    TypeSignatureClass, Volatility,
+};
 use datafusion_macros::user_doc;
 
 #[user_doc(
@@ -62,7 +65,16 @@ impl Default for EndsWithFunc {
 impl EndsWithFunc {
     pub fn new() -> Self {
         Self {
-            signature: Signature::string(2, Volatility::Immutable),
+            signature: Signature::one_of(
+                vec![
+                    TypeSignature::String(2),
+                    TypeSignature::Coercible(vec![
+                        TypeSignatureClass::Native(logical_string()),
+                        TypeSignatureClass::Native(logical_string()),
+                    ]),
+                ],
+                Volatility::Immutable,
+            ),
         }
     }
 }

datafusion/functions/src/string/octet_length.rs

@@ -20,9 +20,12 @@ use arrow::datatypes::DataType;
 use std::any::Any;
 
 use crate::utils::utf8_to_int_type;
+use datafusion_common::types::logical_string;
 use datafusion_common::{exec_err, Result, ScalarValue};
-use datafusion_expr::{ColumnarValue, Documentation, Volatility};
-use datafusion_expr::{ScalarUDFImpl, Signature};
+use datafusion_expr::{
+    ColumnarValue, Documentation, ScalarUDFImpl, Signature, TypeSignature,
+    TypeSignatureClass, Volatility,
+};
 use datafusion_macros::user_doc;
 
 #[user_doc(
@@ -55,7 +58,15 @@ impl Default for OctetLengthFunc {
 impl OctetLengthFunc {
     pub fn new() -> Self {
         Self {
-            signature: Signature::string(1, Volatility::Immutable),
+            signature: Signature::one_of(
+                vec![
+                    TypeSignature::String(1),
+                    TypeSignature::Coercible(vec![TypeSignatureClass::Native(
+                        logical_string(),
+                    )]),
+                ],
+                Volatility::Immutable,
+            ),
         }
     }
 }

datafusion/functions/src/string/starts_with.rs

@@ -23,9 +23,12 @@ use arrow::datatypes::DataType;
 use datafusion_expr::simplify::{ExprSimplifyResult, SimplifyInfo};
 
 use crate::utils::make_scalar_function;
+use datafusion_common::types::logical_string;
 use datafusion_common::{internal_err, Result, ScalarValue};
-use datafusion_expr::{ColumnarValue, Documentation, Expr, Like};
-use datafusion_expr::{ScalarUDFImpl, Signature, Volatility};
+use datafusion_expr::{
+    ColumnarValue, Documentation, Expr, Like, ScalarUDFImpl, Signature, TypeSignature,
+    TypeSignatureClass, Volatility,
+};
 use datafusion_macros::user_doc;
 
 /// Returns true if string starts with prefix.
@@ -64,7 +67,16 @@ impl Default for StartsWithFunc {
 impl StartsWithFunc {
     pub fn new() -> Self {
         Self {
-            signature: Signature::string(2, Volatility::Immutable),
+            signature: Signature::one_of(
+                vec![
+                    TypeSignature::String(2),
+                    TypeSignature::Coercible(vec![
+                        TypeSignatureClass::Native(logical_string()),
+                        TypeSignatureClass::Native(logical_string()),
+                    ]),
+                ],
+                Volatility::Immutable,
+            ),
         }
     }
 }

datafusion/physical-plan/src/values.rs

@@ -34,7 +34,10 @@ use datafusion_execution::TaskContext;
 use datafusion_physical_expr::EquivalenceProperties;
 
 /// Execution plan for values list based relation (produces constant rows)
-#[deprecated(since = "45.0.0", note = "Use `MemoryExec::try_new_as_values` instead")]
+#[deprecated(
+    since = "45.0.0",
+    note = "Use `MemoryExec::try_new_as_values` or `MemoryExec::try_new_from_batches` instead"
+)]
 #[derive(Debug, Clone)]
 pub struct ValuesExec {
     /// The schema