Enhance MIN/MAX integration test and clean up macros

Update the MIN/MAX integration test in basic.rs to use two MemTable
partitions, ensuring the physical plan includes both partial and final
aggregate stages. Retain checks for dictionary-typed output schema
and results. In min_max.rs, remove the no-op min_max_dictionary!
macro and inline the existing generic comparison helper for
improved clarity and efficiency.

Author: kosiew

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

@@ -19,6 +19,7 @@ use super::*;
 use datafusion::common::test_util::batches_to_string;
 use datafusion_catalog::MemTable;
 use datafusion_common::ScalarValue;
+use datafusion_physical_plan::displayable;
 use insta::assert_snapshot;
 
 #[tokio::test]
@@ -444,23 +445,40 @@ async fn count_distinct_dictionary_mixed_values() -> Result<()> {
 
 #[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 ctx = SessionContext::new_with_config(SessionConfig::new().with_target_partitions(2));
 
     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]])?;
+    let batch1 = RecordBatch::try_new(
+        schema.clone(),
+        vec![Arc::new(DictionaryArray::new(
+            Int32Array::from(vec![Some(1), Some(1), None]),
+            Arc::new(StringArray::from(vec!["a", "z", "zz_unused"])),
+        ))],
+    )?;
+    let batch2 = RecordBatch::try_new(
+        schema.clone(),
+        vec![Arc::new(DictionaryArray::new(
+            Int32Array::from(vec![Some(0), Some(1)]),
+            Arc::new(StringArray::from(vec!["a", "d"])),
+        ))],
+    )?;
+    let provider = MemTable::try_new(schema, vec![vec![batch1], vec![batch2]])?;
     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 physical_plan = df.clone().create_physical_plan().await?;
+    let formatted_plan = format!("{}", displayable(physical_plan.as_ref()).indent(true));
+    assert!(formatted_plan.contains("AggregateExec: mode=Partial, gby=[]"));
+    assert!(
+        formatted_plan.contains("AggregateExec: mode=Final, gby=[]")
+            || formatted_plan.contains("AggregateExec: mode=FinalPartitioned, gby=[]")
+    );
+
     let results = df.collect().await?;
 
     assert_eq!(results[0].schema().field(0).data_type(), &dict_type);
@@ -472,7 +490,7 @@ async fn min_max_dictionary_uses_planned_dictionary_path() -> Result<()> {
     +----------+----------+
     | min_dict | max_dict |
     +----------+----------+
-    | z        | z        |
+    | a        | z        |
     +----------+----------+
     "
     );

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

@@ -141,10 +141,6 @@ macro_rules! min_max_generic {
     }};
 }
 
-macro_rules! min_max_dictionary {
-    ($VALUE:expr, $DELTA:expr, $OP:ident) => {{ min_max_generic!($VALUE, $DELTA, $OP) }};
-}
-
 // min/max of two scalar values of the same type
 macro_rules! min_max {
     ($VALUE:expr, $DELTA:expr, $OP:ident) => {{
@@ -423,26 +419,22 @@ macro_rules! min_max {
             ) => {
                 wrap_dictionary_scalar(
                     key_type.as_ref(),
-                    min_max_dictionary!(
-                    lhs_inner.as_ref(),
-                    rhs_inner.as_ref(),
-                    $OP
-                    ),
+                    min_max_generic!(lhs_inner.as_ref(), rhs_inner.as_ref(), $OP),
                 )
             }
 
             (
                 ScalarValue::Dictionary(_, lhs_inner),
                 rhs,
             ) => {
-                min_max_dictionary!(lhs_inner.as_ref(), rhs, $OP)
+                min_max_generic!(lhs_inner.as_ref(), rhs, $OP)
             }
 
             (
                 lhs,
                 ScalarValue::Dictionary(_, rhs_inner),
             ) => {
-                min_max_dictionary!(lhs, rhs_inner.as_ref(), $OP)
+                min_max_generic!(lhs, rhs_inner.as_ref(), $OP)
             }
 
             e => {