feat: support Binary for min/max and regrep_replace

Author: Weijun-H

datafusion/expr/src/aggregate_function.rs

@@ -342,6 +342,7 @@ impl AggregateFunction {
                     .chain(TIMESTAMPS.iter())
                     .chain(DATES.iter())
                     .chain(TIMES.iter())
+                    .chain(BINARYS.iter())
                     .cloned()
                     .collect::<Vec<_>>();
                 Signature::uniform(1, valid, Volatility::Immutable)

datafusion/expr/src/built_in_function.rs

@@ -1086,6 +1086,13 @@ impl BuiltinScalarFunction {
             BuiltinScalarFunction::RegexpReplace => Signature::one_of(
                 vec![
                     Exact(vec![Utf8, Utf8, Utf8]),
+                    Exact(vec![Binary, Utf8, Utf8]),
+                    Exact(vec![Utf8, Binary, Utf8]),
+                    Exact(vec![Utf8, Utf8, Binary]),
+                    Exact(vec![Binary, Binary, Utf8]),
+                    Exact(vec![Binary, Utf8, Binary]),
+                    Exact(vec![Utf8, Binary, Binary]),
+                    Exact(vec![Binary, Binary, Binary]),
                     Exact(vec![Utf8, Utf8, Utf8, Utf8]),
                 ],
                 self.volatility(),
@@ -1392,8 +1399,8 @@ macro_rules! make_utf8_to_return_type {
     ($FUNC:ident, $largeUtf8Type:expr, $utf8Type:expr) => {
         fn $FUNC(arg_type: &DataType, name: &str) -> Result<DataType> {
             Ok(match arg_type {
-                DataType::LargeUtf8 => $largeUtf8Type,
-                DataType::Utf8 => $utf8Type,
+                DataType::LargeUtf8 | DataType::LargeBinary=> $largeUtf8Type,
+                DataType::Utf8 | DataType::Binary => $utf8Type,
                 DataType::Null => DataType::Null,
                 DataType::Dictionary(_, value_type) => match **value_type {
                     DataType::LargeUtf8 => $largeUtf8Type,

datafusion/expr/src/type_coercion/aggregates.rs

@@ -75,6 +75,8 @@ pub static TIMESTAMPS: &[DataType] = &[
 
 pub static DATES: &[DataType] = &[DataType::Date32, DataType::Date64];
 
+pub static BINARYS: &[DataType] = &[DataType::Binary, DataType::LargeBinary];
+
 pub static TIMES: &[DataType] = &[
     DataType::Time32(TimeUnit::Second),
     DataType::Time32(TimeUnit::Millisecond),

datafusion/physical-expr/src/aggregate/min_max.rs

@@ -31,12 +31,12 @@ use arrow::datatypes::{
 };
 use arrow::{
     array::{
-        ArrayRef, BooleanArray, Date32Array, Date64Array, Float32Array, Float64Array,
-        Int16Array, Int32Array, Int64Array, Int8Array, LargeStringArray, StringArray,
-        Time32MillisecondArray, Time32SecondArray, Time64MicrosecondArray,
-        Time64NanosecondArray, TimestampMicrosecondArray, TimestampMillisecondArray,
-        TimestampNanosecondArray, TimestampSecondArray, UInt16Array, UInt32Array,
-        UInt64Array, UInt8Array,
+        ArrayRef, BinaryArray, BooleanArray, Date32Array, Date64Array, Float32Array,
+        Float64Array, Int16Array, Int32Array, Int64Array, Int8Array, LargeBinaryArray,
+        LargeStringArray, StringArray, Time32MillisecondArray, Time32SecondArray,
+        Time64MicrosecondArray, Time64NanosecondArray, TimestampMicrosecondArray,
+        TimestampMillisecondArray, TimestampNanosecondArray, TimestampSecondArray,
+        UInt16Array, UInt32Array, UInt64Array, UInt8Array,
     },
     datatypes::Field,
 };

datafusion/physical-expr/src/functions.rs

@@ -690,6 +690,21 @@ pub fn create_physical_fun(
                     let func = specializer_func(args)?;
                     func(args)
                 }
+                DataType::Binary => {
+                    let args: Vec<ColumnarValue> = args
+                        .iter()
+                        .map(|col_value| {
+                            cast_column(col_value, &DataType::Utf8, None).unwrap()
+                        })
+                        .collect();
+                    let specializer_func = invoke_on_columnar_value_if_regex_expressions_feature_flag!(
+                        specialize_regexp_replace,
+                        i32,
+                        "regexp_replace"
+                    );
+                    let func = specializer_func(&args)?;
+                    func(&args)
+                }
                 other => internal_err!(
                     "Unsupported data type {other:?} for function regexp_replace"
                 ),

datafusion/sqllogictest/test_files/aggregate.slt

@@ -1976,7 +1976,6 @@ SELECT
   count(largebinary)
 FROM t
 
-
 # with groupby
 query error DataFusion error: External error: Internal error: Min/Max accumulator not implemented for type Binary\. This was likely caused by a bug in DataFusion's code and we would welcome that you file an bug report in our issue tracker
 SELECT