Fix Coalesce casting logic to follows what Postgres and DuckDB do. Introduce signature that do non-comparison coercion

datafusion/expr/src/signature.rs

@@ -92,14 +92,19 @@ pub enum TypeSignature {
     /// A function such as `concat` is `Variadic(vec![DataType::Utf8, DataType::LargeUtf8])`
     Variadic(Vec<DataType>),
     /// One or more arguments of an arbitrary but equal type.
-    /// DataFusion attempts to coerce all argument types to match the first argument's type
+    /// DataFusion attempts to coerce all argument types to match to the common type with comparision coercion.
     ///
     /// # Examples
     /// Given types in signature should be coercible to the same final type.
     /// A function such as `make_array` is `VariadicEqual`.
     ///
     /// `make_array(i32, i64) -> make_array(i64, i64)`
     VariadicEqual,
+    /// One or more arguments of an arbitrary but equal type or Null.
+    /// No coercion is attempted.
+    ///
+    /// Functions like `coalesce` is `VariadicEqual`.
+    VariadicEqualOrNull,
     /// One or more arguments with arbitrary types
     VariadicAny,
     /// Fixed number of arguments of an arbitrary but equal type out of a list of valid types.
@@ -193,6 +198,9 @@ impl TypeSignature {
             TypeSignature::VariadicEqual => {
                 vec!["CoercibleT, .., CoercibleT".to_string()]
             }
+            TypeSignature::VariadicEqualOrNull => {
+                vec!["T or Null, .., T or Null".to_string()]
+            }
             TypeSignature::VariadicAny => vec!["Any, .., Any".to_string()],
             TypeSignature::OneOf(sigs) => {
                 sigs.iter().flat_map(|s| s.to_string_repr()).collect()
@@ -226,6 +234,11 @@ impl TypeSignature {
             _ => false,
         }
     }
+
+    /// Skip coercion to match the signature.
+    pub(crate) fn skip_coercion(&self) -> bool {
+        matches!(self, TypeSignature::VariadicEqualOrNull)
+    }
 }
 
 /// Defines the supported argument types ([`TypeSignature`]) and [`Volatility`] for a function.
@@ -255,13 +268,20 @@ impl Signature {
             volatility,
         }
     }
-    /// An arbitrary number of arguments of the same type.
+    /// One or more number of arguments to the same type.
     pub fn variadic_equal(volatility: Volatility) -> Self {
         Self {
             type_signature: TypeSignature::VariadicEqual,
             volatility,
         }
     }
+    /// One or more number of arguments of the same type.
+    pub fn variadic_equal_or_null(volatility: Volatility) -> Self {
+        Self {
+            type_signature: TypeSignature::VariadicEqualOrNull,
+            volatility,
+        }
+    }
     /// An arbitrary number of arguments of any type.
     pub fn variadic_any(volatility: Volatility) -> Self {
         Self {

datafusion/expr/src/type_coercion/functions.rs

@@ -28,7 +28,7 @@ use arrow::{
 use datafusion_common::utils::{coerced_fixed_size_list_to_list, list_ndims};
 use datafusion_common::{internal_datafusion_err, internal_err, plan_err, Result};
 
-use super::binary::{comparison_binary_numeric_coercion, comparison_coercion};
+use super::binary::comparison_coercion;
 
 /// Performs type coercion for function arguments.
 ///
@@ -62,11 +62,13 @@ pub fn data_types(
         return Ok(current_types.to_vec());
     }
 
-    // Try and coerce the argument types to match the signature, returning the
-    // coerced types from the first matching signature.
-    for valid_types in valid_types {
-        if let Some(types) = maybe_data_types(&valid_types, current_types) {
-            return Ok(types);
+    if !signature.type_signature.skip_coercion() {
+        // Try and coerce the argument types to match the signature, returning the
+        // coerced types from the first matching signature.
+        for valid_types in valid_types {
+            if let Some(types) = maybe_data_types(&valid_types, current_types) {
+                return Ok(types);
+            }
         }
     }
 
@@ -184,6 +186,27 @@ fn get_valid_types(
             .iter()
             .map(|valid_type| (0..*number).map(|_| valid_type.clone()).collect())
             .collect(),
+        TypeSignature::VariadicEqualOrNull => {
+            current_types
+                .iter()
+                .find(|&t| t != &DataType::Null)
+                .map_or_else(
+                    || vec![vec![DataType::Null; current_types.len()]],
+                    |t| {
+                        let valid_types = current_types
+                            .iter()
+                            .map(|d| {
+                                if d != &DataType::Null {
+                                    t.clone()
+                                } else {
+                                    DataType::Null
+                                }
+                            })
+                            .collect::<Vec<_>>();
+                        vec![valid_types]
+                    },
+                )
+        }
         TypeSignature::VariadicEqual => {
             let new_type = current_types.iter().skip(1).try_fold(
                 current_types.first().unwrap().clone(),
@@ -450,19 +473,7 @@ fn coerced_from<'a>(
         {
             Some(type_into.clone())
         }
-        // More coerce rules.
-        // Note that not all rules in `comparison_coercion` can be reused here.
-        // For example, all numeric types can be coerced into Utf8 for comparison,
-        // but not for function arguments.
-        _ => comparison_binary_numeric_coercion(type_into, type_from).and_then(
-            |coerced_type| {
-                if *type_into == coerced_type {
-                    Some(coerced_type)
-                } else {
-                    None
-                }
-            },
-        ),
+        _ => None,
     }
 }
 

datafusion/functions/src/core/coalesce.rs

@@ -23,7 +23,6 @@ use arrow::compute::{and, is_not_null, is_null};
 use arrow::datatypes::DataType;
 
 use datafusion_common::{exec_err, Result};
-use datafusion_expr::type_coercion::functions::data_types;
 use datafusion_expr::ColumnarValue;
 use datafusion_expr::{ScalarUDFImpl, Signature, Volatility};
 
@@ -41,7 +40,7 @@ impl Default for CoalesceFunc {
 impl CoalesceFunc {
     pub fn new() -> Self {
         Self {
-            signature: Signature::variadic_equal(Volatility::Immutable),
+            signature: Signature::variadic_equal_or_null(Volatility::Immutable),
         }
     }
 }
@@ -60,9 +59,11 @@ impl ScalarUDFImpl for CoalesceFunc {
     }
 
     fn return_type(&self, arg_types: &[DataType]) -> Result<DataType> {
-        // COALESCE has multiple args and they might get coerced, get a preview of this
-        let coerced_types = data_types(arg_types, self.signature());
-        coerced_types.map(|types| types[0].clone())
+        Ok(arg_types
+            .iter()
+            .find(|&t| t != &DataType::Null)
+            .unwrap_or(&DataType::Null)
+            .clone())
     }
 
     /// coalesce evaluates to the first value which is not NULL

datafusion/sqllogictest/test_files/expr.slt

@@ -1897,7 +1897,7 @@ SELECT substring('alphabet' for 1);
 ----
 a
 
-# The 'from' and 'for' parameters don't support string types, because they should be treated as 
+# The 'from' and 'for' parameters don't support string types, because they should be treated as
 # regular expressions, which we have not implemented yet.
 query error DataFusion error: Error during planning: No function matches the given name and argument types
 SELECT substring('alphabet' FROM '3')