Refactor min/max logic for shared row-wise handling

Consolidate row-wise min/max scan logic into a single helper
in min_max.rs to ensure consistency between dictionary
and generic complex-type paths. Add regression test for
the float dictionary handling NaN and -inf cases,
validating ordering semantics across batches.

Author: kosiew

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

@@ -455,10 +455,7 @@ macro_rules! min_max {
     }};
 }
 
-fn dictionary_batch_extreme(
-    values: &ArrayRef,
-    ordering: Ordering,
-) -> Result<ScalarValue> {
+fn scalar_batch_extreme(values: &ArrayRef, ordering: Ordering) -> Result<ScalarValue> {
     let mut extreme: Option<ScalarValue> = None;
 
     for i in 0..values.len() {
@@ -823,32 +820,14 @@ pub fn min_batch(values: &ArrayRef) -> Result<ScalarValue> {
         DataType::FixedSizeList(_, _) => {
             min_max_batch_generic(values, Ordering::Greater)?
         }
-        DataType::Dictionary(_, _) => {
-            dictionary_batch_extreme(values, Ordering::Greater)?
-        }
+        DataType::Dictionary(_, _) => scalar_batch_extreme(values, Ordering::Greater)?,
         _ => min_max_batch!(values, min),
     })
 }
 
 /// Generic min/max implementation for complex types
 fn min_max_batch_generic(array: &ArrayRef, ordering: Ordering) -> Result<ScalarValue> {
-    let mut non_null_indices = (0..array.len()).filter(|&i| !array.is_null(i));
-    let Some(first_idx) = non_null_indices.next() else {
-        return ScalarValue::try_from(array.data_type());
-    };
-
-    let mut extreme = ScalarValue::try_from_array(array, first_idx)?;
-    for i in non_null_indices {
-        let current = ScalarValue::try_from_array(array, i)?;
-        if current.is_null() {
-            continue;
-        }
-        if extreme.is_null() || extreme.try_cmp(&current)? == ordering {
-            extreme = current;
-        }
-    }
-
-    Ok(extreme)
+    scalar_batch_extreme(array, ordering)
 }
 
 /// dynamically-typed max(array) -> ScalarValue
@@ -900,7 +879,7 @@ pub fn max_batch(values: &ArrayRef) -> Result<ScalarValue> {
         DataType::List(_) => min_max_batch_generic(values, Ordering::Less)?,
         DataType::LargeList(_) => min_max_batch_generic(values, Ordering::Less)?,
         DataType::FixedSizeList(_, _) => min_max_batch_generic(values, Ordering::Less)?,
-        DataType::Dictionary(_, _) => dictionary_batch_extreme(values, Ordering::Less)?,
+        DataType::Dictionary(_, _) => scalar_batch_extreme(values, Ordering::Less)?,
         _ => min_max_batch!(values, max),
     })
 }

datafusion/functions-aggregate/src/min_max.rs

@@ -1288,6 +1288,13 @@ mod tests {
         ) as ArrayRef
     }
 
+    fn float_dictionary_batch(values: &[f32], keys: &[Option<i32>]) -> ArrayRef {
+        let values = Arc::new(Float32Array::from(values.to_vec())) as ArrayRef;
+        Arc::new(
+            DictionaryArray::try_new(Int32Array::from(keys.to_vec()), values).unwrap(),
+        ) as ArrayRef
+    }
+
     fn evaluate_dictionary_accumulator(
         mut acc: impl Accumulator,
         batches: &[ArrayRef],
@@ -1377,4 +1384,35 @@ mod tests {
 
         assert_dictionary_min_max(&dict_type, &[batch1, batch2], "a", "d")
     }
+
+    #[test]
+    fn test_min_max_dictionary_float_with_nans() -> Result<()> {
+        let dict_type =
+            DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Float32));
+        let batch1 = float_dictionary_batch(&[0.0, f32::NAN], &[Some(0), Some(1)]);
+        let batch2 = float_dictionary_batch(&[f32::NEG_INFINITY], &[Some(0)]);
+
+        let min_result = evaluate_dictionary_accumulator(
+            MinAccumulator::try_new(&dict_type)?,
+            &[Arc::clone(&batch1), Arc::clone(&batch2)],
+        )?;
+        assert_eq!(
+            min_result,
+            dict_scalar(
+                DataType::Int32,
+                ScalarValue::Float32(Some(f32::NEG_INFINITY)),
+            )
+        );
+
+        let max_result = evaluate_dictionary_accumulator(
+            MaxAccumulator::try_new(&dict_type)?,
+            &[batch1, batch2],
+        )?;
+        assert_eq!(
+            max_result,
+            dict_scalar(DataType::Int32, ScalarValue::Float32(Some(f32::NAN)))
+        );
+
+        Ok(())
+    }
 }