Introduce TypeSignatureClass::Binary to allow accepting arbitrarily sized FixedSizeBinary arguments (#17531)

* Introduce wildcard const for FixedSizeBinary type signature

* Add Binary to TypeSignatureClass

* Remove FIXED_SIZE_BINARY_WILDCARD

Author: Jefffrey

datafusion/common/src/types/native.rs

@@ -472,4 +472,9 @@ impl NativeType {
     pub fn is_duration(&self) -> bool {
         matches!(self, NativeType::Duration(_))
     }
+
+    #[inline]
+    pub fn is_binary(&self) -> bool {
+        matches!(self, NativeType::Binary | NativeType::FixedSizeBinary(_))
+    }
 }

datafusion/expr-common/src/signature.rs

@@ -266,6 +266,8 @@ pub enum TypeSignatureClass {
     // TODO:
     // Numeric
     Integer,
+    /// Encompasses both the native Binary as well as arbitrarily sized FixedSizeBinary types
+    Binary,
 }
 
 impl Display for TypeSignatureClass {
@@ -303,6 +305,9 @@ impl TypeSignatureClass {
             TypeSignatureClass::Integer => {
                 vec![DataType::Int64]
             }
+            TypeSignatureClass::Binary => {
+                vec![DataType::Binary]
+            }
         }
     }
 
@@ -322,6 +327,7 @@ impl TypeSignatureClass {
             TypeSignatureClass::Interval if logical_type.is_interval() => true,
             TypeSignatureClass::Duration if logical_type.is_duration() => true,
             TypeSignatureClass::Integer if logical_type.is_integer() => true,
+            TypeSignatureClass::Binary if logical_type.is_binary() => true,
             _ => false,
         }
     }
@@ -352,6 +358,9 @@ impl TypeSignatureClass {
             TypeSignatureClass::Integer if native_type.is_integer() => {
                 Ok(origin_type.to_owned())
             }
+            TypeSignatureClass::Binary if native_type.is_binary() => {
+                Ok(origin_type.to_owned())
+            }
             _ => internal_err!("May miss the matching logic in `matches_native_type`"),
         }
     }

datafusion/spark/src/function/bitmap/bitmap_count.rs

@@ -27,9 +27,10 @@ use arrow::datatypes::DataType::{
     Binary, BinaryView, FixedSizeBinary, Int64, LargeBinary,
 };
 use datafusion_common::utils::take_function_args;
-use datafusion_common::{internal_datafusion_err, internal_err, plan_err, Result};
+use datafusion_common::{internal_datafusion_err, internal_err, Result};
 use datafusion_expr::{
-    ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility,
+    Coercion, ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature,
+    TypeSignatureClass, Volatility,
 };
 use datafusion_functions::utils::make_scalar_function;
 use datafusion_functions::{downcast_arg, downcast_named_arg};
@@ -48,8 +49,10 @@ impl Default for BitmapCount {
 impl BitmapCount {
     pub fn new() -> Self {
         Self {
-            // TODO: add definitive TypeSignature after https://github.com/apache/datafusion/issues/17291 is done
-            signature: Signature::any(1, Volatility::Immutable),
+            signature: Signature::coercible(
+                vec![Coercion::new_exact(TypeSignatureClass::Binary)],
+                Volatility::Immutable,
+            ),
         }
     }
 }
@@ -67,15 +70,8 @@ impl ScalarUDFImpl for BitmapCount {
         &self.signature
     }
 
-    fn return_type(&self, arg_types: &[DataType]) -> Result<DataType> {
-        match arg_types.first() {
-            Some(Binary | BinaryView | FixedSizeBinary(_) | LargeBinary) => Ok(Int64),
-            Some(data_type) => plan_err!(
-                "bitmap_count expects Binary/BinaryView/FixedSizeBinary/LargeBinary as argument, got {:?}",
-                data_type
-            ),
-            None => internal_err!("bitmap_count does not support zero arguments"),
-        }
+    fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+        Ok(Int64)
     }
 
     fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {