Revert to dad6e02: Refactor dictionary handling and simplify batch logic

Author: kosiew

datafusion/functions-aggregate-common/src/min_max.rs

@@ -141,10 +141,7 @@ macro_rules! min_max_generic {
     }};
 }
 
-// min/max of two logically compatible scalar values.
-// Dictionary scalars are unwrapped to their inner values for comparison,
-// then rewrapped with the dictionary key type when both inputs are dictionaries
-// after validating that their key types match.
+// min/max of two scalar values of the same type
 macro_rules! min_max {
     ($VALUE:expr, $DELTA:expr, $OP:ident) => {{
         Ok(match ($VALUE, $DELTA) {
@@ -425,40 +422,24 @@ macro_rules! min_max {
                 let result = min_max_generic!(lhs, rhs, $OP);
 
                 match lhs_key_type.zip(rhs_key_type) {
-                    Some((lhs_key_type, rhs_key_type)) => {
-                        if lhs_key_type != rhs_key_type {
-                            return internal_err!(
-                                "MIN/MAX is not expected to receive dictionary scalars with different key types ({:?} vs {:?})",
-                                lhs_key_type,
-                                rhs_key_type
-                            );
-                        }
-
-                        ScalarValue::Dictionary(
-                            Box::new(lhs_key_type.clone()),
-                            Box::new(result),
-                        )
+                    Some((key_type, _)) => {
+                        ScalarValue::Dictionary(Box::new(key_type.clone()), Box::new(result))
                     }
                     None => result,
                 }
             }
 
             e => {
                 return internal_err!(
-                    "MIN/MAX is not expected to receive logically incompatible scalar values {:?}",
+                    "MIN/MAX is not expected to receive scalars of incompatible types {:?}",
                     e
                 )
             }
         })
     }};
 }
 
-/// Finds the min/max by scanning logical rows via `ScalarValue::try_from_array`.
-///
-/// This path is required for dictionary arrays because comparing
-/// `dictionary.values()` is not semantically correct: it can include
-/// unreferenced values and ignore null key positions.
-fn scalar_row_extreme(values: &ArrayRef, ordering: Ordering) -> Result<ScalarValue> {
+fn scalar_batch_extreme(values: &ArrayRef, ordering: Ordering) -> Result<ScalarValue> {
     let mut index = 0;
     let mut extreme = loop {
         if index == values.len() {
@@ -494,9 +475,7 @@ fn dictionary_scalar_parts(value: &ScalarValue) -> (&ScalarValue, Option<&DataTy
     }
 }
 
-// Primitive, string, and binary types use specialized Arrow min/max kernels.
-// These remaining types fall back to scalar row-by-row logical comparison.
-fn requires_logical_row_scan(data_type: &DataType) -> bool {
+fn is_row_wise_batch_type(data_type: &DataType) -> bool {
     matches!(
         data_type,
         DataType::Struct(_)
@@ -844,8 +823,8 @@ pub fn min_batch(values: &ArrayRef) -> Result<ScalarValue> {
                 min_binary_view
             )
         }
-        data_type if requires_logical_row_scan(data_type) => {
-            scalar_row_extreme(values, Ordering::Greater)?
+        data_type if is_row_wise_batch_type(data_type) => {
+            scalar_batch_extreme(values, Ordering::Greater)?
         }
         _ => min_max_batch!(values, min),
     })
@@ -896,8 +875,8 @@ pub fn max_batch(values: &ArrayRef) -> Result<ScalarValue> {
             let value = value.map(|e| e.to_vec());
             ScalarValue::FixedSizeBinary(*size, value)
         }
-        data_type if requires_logical_row_scan(data_type) => {
-            scalar_row_extreme(values, Ordering::Less)?
+        data_type if is_row_wise_batch_type(data_type) => {
+            scalar_batch_extreme(values, Ordering::Less)?
         }
         _ => min_max_batch!(values, max),
     })