Fix dictionary min/max behavior in DataFusion

Update min_max.rs to ensure dictionary batches iterate
actual array rows, comparing referenced scalar values.
Unreferenced dictionary entries no longer affect MIN/MAX,
and referenced null values are correctly skipped.
Expanded tests to cover these changes and updated
expectations
Added regression tests for unreferenced and referenced
null dictionary values.

Author: kosiew

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

@@ -18,8 +18,8 @@
 //! Basic min/max functionality shared across DataFusion aggregate functions
 
 use arrow::array::{
-    ArrayRef, AsArray as _, BinaryArray, BinaryViewArray, BooleanArray, Date32Array,
-    Date64Array, Decimal32Array, Decimal64Array, Decimal128Array, Decimal256Array,
+    ArrayRef, BinaryArray, BinaryViewArray, BooleanArray, Date32Array, Date64Array,
+    Decimal32Array, Decimal64Array, Decimal128Array, Decimal256Array,
     DurationMicrosecondArray, DurationMillisecondArray, DurationNanosecondArray,
     DurationSecondArray, FixedSizeBinaryArray, Float16Array, Float32Array, Float64Array,
     Int8Array, Int16Array, Int32Array, Int64Array, IntervalDayTimeArray,
@@ -457,13 +457,23 @@ macro_rules! min_max {
 
 fn dictionary_batch_extreme(
     values: &ArrayRef,
-    extreme_fn: fn(&ArrayRef) -> Result<ScalarValue>,
+    ordering: Ordering,
 ) -> Result<ScalarValue> {
-    let DataType::Dictionary(key_type, _) = values.data_type() else {
-        unreachable!("dictionary_batch_extreme requires dictionary arrays")
-    };
-    let inner = extreme_fn(values.as_any_dictionary().values())?;
-    Ok(wrap_dictionary_scalar(key_type.as_ref(), inner))
+    let mut extreme: Option<ScalarValue> = None;
+
+    for i in 0..values.len() {
+        let current = ScalarValue::try_from_array(values, i)?;
+        if current.is_null() {
+            continue;
+        }
+
+        match &extreme {
+            Some(existing) if existing.try_cmp(&current)? != ordering => {}
+            _ => extreme = Some(current),
+        }
+    }
+
+    extreme.map_or_else(|| ScalarValue::try_from(values.data_type()), Ok)
 }
 
 fn wrap_dictionary_scalar(key_type: &DataType, value: ScalarValue) -> ScalarValue {
@@ -813,7 +823,9 @@ pub fn min_batch(values: &ArrayRef) -> Result<ScalarValue> {
         DataType::FixedSizeList(_, _) => {
             min_max_batch_generic(values, Ordering::Greater)?
         }
-        DataType::Dictionary(_, _) => dictionary_batch_extreme(values, min_batch)?,
+        DataType::Dictionary(_, _) => {
+            dictionary_batch_extreme(values, Ordering::Greater)?
+        }
         _ => min_max_batch!(values, min),
     })
 }
@@ -828,7 +840,10 @@ fn min_max_batch_generic(array: &ArrayRef, ordering: Ordering) -> Result<ScalarV
     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 extreme.try_cmp(&current)? == ordering {
+        if current.is_null() {
+            continue;
+        }
+        if extreme.is_null() || extreme.try_cmp(&current)? == ordering {
             extreme = current;
         }
     }
@@ -885,7 +900,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, max_batch)?,
+        DataType::Dictionary(_, _) => dictionary_batch_extreme(values, Ordering::Less)?,
         _ => min_max_batch!(values, max),
     })
 }

datafusion/functions-aggregate/src/min_max.rs

@@ -1259,7 +1259,7 @@ mod tests {
         let mut max_acc = MaxAccumulator::try_new(&rt_type)?;
         max_acc.update_batch(&[Arc::clone(&dict_array_ref)])?;
         let max_result = max_acc.evaluate()?;
-        assert_eq!(max_result, ScalarValue::Utf8(Some("🦀".to_string())));
+        assert_eq!(max_result, ScalarValue::Utf8(Some("d".to_string())));
         Ok(())
     }
 
@@ -1278,6 +1278,16 @@ mod tests {
         ) as ArrayRef
     }
 
+    fn optional_string_dictionary_batch(
+        values: &[Option<&str>],
+        keys: &[Option<i32>],
+    ) -> ArrayRef {
+        let values = Arc::new(StringArray::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],
@@ -1336,6 +1346,28 @@ mod tests {
         assert_dictionary_min_max(&dict_type, &[dict_array_ref], "a", "c")
     }
 
+    #[test]
+    fn test_min_max_dictionary_ignores_unreferenced_values() -> Result<()> {
+        let dict_array_ref = string_dictionary_batch(
+            &["a", "z", "zz_unused"],
+            &[Some(1), Some(1), None],
+        );
+        let dict_type = dict_array_ref.data_type().clone();
+
+        assert_dictionary_min_max(&dict_type, &[dict_array_ref], "z", "z")
+    }
+
+    #[test]
+    fn test_min_max_dictionary_ignores_referenced_null_values() -> Result<()> {
+        let dict_array_ref = optional_string_dictionary_batch(
+            &[Some("b"), None, Some("a"), Some("d")],
+            &[Some(0), Some(1), Some(2), Some(3)],
+        );
+        let dict_type = dict_array_ref.data_type().clone();
+
+        assert_dictionary_min_max(&dict_type, &[dict_array_ref], "a", "d")
+    }
+
     #[test]
     fn test_min_max_dictionary_multi_batch() -> Result<()> {
         let dict_type =