Fix MIN/MAX to preserve dictionary types in execution

Update get_min_max_result_type to maintain Dictionary<K, V>
instead of unwrapping to V, allowing planned MIN/MAX execution to
utilize the dictionary-aware accumulator. Add end-to-end SQL
regression test to ensure MIN/MAX properly ignores unreferenced
dictionary values and preserves dictionary-typed output schema.
Adjust unit expectations for dictionary coercion tests to reflect
new planned-path behavior.

Author: kosiew

datafusion/core/tests/sql/aggregates/basic.rs

@@ -442,6 +442,44 @@ async fn count_distinct_dictionary_mixed_values() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn min_max_dictionary_uses_planned_dictionary_path() -> Result<()> {
+    let ctx = SessionContext::new();
+
+    let dict_values = StringArray::from(vec!["a", "z", "zz_unused"]);
+    let dict_indices = Int32Array::from(vec![Some(1), Some(1), None]);
+    let dict = DictionaryArray::new(dict_indices, Arc::new(dict_values));
+
+    let dict_type =
+        DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8));
+    let schema = Arc::new(Schema::new(vec![Field::new("dict", dict_type.clone(), true)]));
+
+    let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(dict)])?;
+    let provider = MemTable::try_new(schema, vec![vec![batch]])?;
+    ctx.register_table("t", Arc::new(provider))?;
+
+    let df = ctx
+        .sql("SELECT min(dict) AS min_dict, max(dict) AS max_dict FROM t")
+        .await?;
+    let results = df.collect().await?;
+
+    assert_eq!(results[0].schema().field(0).data_type(), &dict_type);
+    assert_eq!(results[0].schema().field(1).data_type(), &dict_type);
+
+    assert_snapshot!(
+        batches_to_string(&results),
+        @r"
+    +----------+----------+
+    | min_dict | max_dict |
+    +----------+----------+
+    | z        | z        |
+    +----------+----------+
+    "
+    );
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn group_by_ree_dict_column() -> Result<()> {
     let ctx = SessionContext::new();

datafusion/functions-aggregate/src/min_max.rs

@@ -53,7 +53,6 @@ use datafusion_expr::{GroupsAccumulator, StatisticsArgs};
 use datafusion_macros::user_doc;
 use half::f16;
 use std::mem::size_of_val;
-use std::ops::Deref;
 
 fn get_min_max_result_type(input_types: &[DataType]) -> Result<Vec<DataType>> {
     // make sure that the input types only has one element.
@@ -63,17 +62,12 @@ fn get_min_max_result_type(input_types: &[DataType]) -> Result<Vec<DataType>> {
             input_types.len()
         );
     }
-    // min and max support the dictionary data type
-    // unpack the dictionary to get the value
-    match &input_types[0] {
-        DataType::Dictionary(_, dict_value_type) => {
-            // TODO add checker, if the value type is complex data type
-            Ok(vec![dict_value_type.deref().clone()])
-        }
-        // TODO add checker for datatype which min and max supported
-        // For example, the `Struct` and `Map` type are not supported in the MIN and MAX function
-        _ => Ok(input_types.to_vec()),
-    }
+    // Preserve dictionary inputs so planned MIN/MAX execution uses the same
+    // dictionary-aware accumulator/state path as direct accumulator tests.
+    //
+    // TODO add checker for datatype which min and max supported.
+    // For example, the `Struct` and `Map` type are not supported in the MIN and MAX function.
+    Ok(input_types.to_vec())
 }
 
 #[user_doc(
@@ -1223,6 +1217,10 @@ mod tests {
             vec![DataType::Decimal128(10, 2)],
             vec![DataType::Decimal256(1, 1)],
             vec![DataType::Utf8],
+            vec![DataType::Dictionary(
+                Box::new(DataType::Int32),
+                Box::new(DataType::Utf8),
+            )],
         ];
         for fun in funs {
             for input_type in &input_types {
@@ -1237,7 +1235,13 @@ mod tests {
         let data_type =
             DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8));
         let result = get_min_max_result_type(&[data_type])?;
-        assert_eq!(result, vec![DataType::Utf8]);
+        assert_eq!(
+            result,
+            vec![DataType::Dictionary(
+                Box::new(DataType::Int32),
+                Box::new(DataType::Utf8),
+            )]
+        );
         Ok(())
     }
 
@@ -1254,12 +1258,18 @@ mod tests {
         let mut min_acc = MinAccumulator::try_new(&rt_type)?;
         min_acc.update_batch(&[Arc::clone(&dict_array_ref)])?;
         let min_result = min_acc.evaluate()?;
-        assert_eq!(min_result, ScalarValue::Utf8(Some("a".to_string())));
+        assert_eq!(
+            min_result,
+            dict_scalar(DataType::Int32, ScalarValue::Utf8(Some("a".to_string())))
+        );
 
         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("d".to_string())));
+        assert_eq!(
+            max_result,
+            dict_scalar(DataType::Int32, ScalarValue::Utf8(Some("d".to_string())))
+        );
         Ok(())
     }