Refactor min/max batch handling and simplify JSON logic

Collapse repeated complex-type match arms in min_batch and
max_batch. Introduce a shared update_extreme path for the
min/max accumulators. Simplify the ordered_keys function
and factor the repeated pretty-print container logic
into write_delimited_entries. Use a local macro for JSON
stringification deduplication in display.rs to maintain
behavior without adding dependencies.

Author: kosiew

datafusion/expr/src/logical_plan/display.rs

@@ -295,33 +295,69 @@ fn ordered_keys(map: &serde_json::Map<String, serde_json::Value>) -> Vec<&str> {
         return keys;
     }
 
-    let mut ordered_keys = Vec::with_capacity(keys.len());
+    let mut remaining_keys = Vec::with_capacity(keys.len().saturating_sub(1));
+    let mut has_plans_key = false;
+    let mut has_output_key = false;
+    for key in keys {
+        match key {
+            NODE_TYPE_KEY => {}
+            PLANS_KEY => has_plans_key = true,
+            OUTPUT_KEY => has_output_key = true,
+            _ => remaining_keys.push(key),
+        }
+    }
+
+    let mut ordered_keys = Vec::with_capacity(map.len());
     ordered_keys.push(NODE_TYPE_KEY);
-    ordered_keys.extend(
-        keys.iter()
-            .copied()
-            .filter(|key| !matches!(*key, NODE_TYPE_KEY | PLANS_KEY | OUTPUT_KEY)),
-    );
-    ordered_keys.extend(
-        [PLANS_KEY, OUTPUT_KEY]
-            .into_iter()
-            .filter(|key| map.contains_key(*key)),
-    );
+    ordered_keys.extend(remaining_keys);
+    if has_plans_key {
+        ordered_keys.push(PLANS_KEY);
+    }
+    if has_output_key {
+        ordered_keys.push(OUTPUT_KEY);
+    }
     ordered_keys
 }
 
-fn json_to_string(value: &serde_json::Value) -> datafusion_common::Result<String> {
-    serde_json::to_string(value).map_err(|e| DataFusionError::External(Box::new(e)))
-}
-
-fn json_key_to_string(key: &str) -> datafusion_common::Result<String> {
-    serde_json::to_string(key).map_err(|e| DataFusionError::External(Box::new(e)))
+macro_rules! to_json_string {
+    ($value:expr) => {
+        serde_json::to_string($value).map_err(|e| DataFusionError::External(Box::new(e)))
+    };
 }
 
 fn push_indent(buf: &mut String, indent: usize) {
     buf.push_str(&" ".repeat(indent * 2));
 }
 
+fn write_delimited_entries(
+    buf: &mut String,
+    indent: usize,
+    open: char,
+    close: char,
+    len: usize,
+    mut write_entry: impl FnMut(usize, &mut String) -> datafusion_common::Result<()>,
+) -> datafusion_common::Result<()> {
+    if len == 0 {
+        buf.push(open);
+        buf.push(close);
+        return Ok(());
+    }
+
+    buf.push(open);
+    buf.push('\n');
+    for idx in 0..len {
+        push_indent(buf, indent + 1);
+        write_entry(idx, buf)?;
+        if idx + 1 != len {
+            buf.push(',');
+        }
+        buf.push('\n');
+    }
+    push_indent(buf, indent);
+    buf.push(close);
+    Ok(())
+}
+
 fn write_ordered_json(
     value: &serde_json::Value,
     buf: &mut String,
@@ -332,55 +368,25 @@ fn write_ordered_json(
         | serde_json::Value::Bool(_)
         | serde_json::Value::Number(_)
         | serde_json::Value::String(_) => {
-            buf.push_str(&json_to_string(value)?);
+            buf.push_str(&to_json_string!(value)?);
         }
         serde_json::Value::Array(values) => {
-            if values.is_empty() {
-                buf.push_str("[]");
-                return Ok(());
-            }
-
-            buf.push('[');
-            buf.push('\n');
-            for (idx, value) in values.iter().enumerate() {
-                push_indent(buf, indent + 1);
-                write_ordered_json(value, buf, indent + 1)?;
-                if idx + 1 != values.len() {
-                    buf.push(',');
-                }
-                buf.push('\n');
-            }
-            push_indent(buf, indent);
-            buf.push(']');
+            write_delimited_entries(buf, indent, '[', ']', values.len(), |idx, buf| {
+                write_ordered_json(&values[idx], buf, indent + 1)
+            })?;
         }
         serde_json::Value::Object(map) => {
-            if map.is_empty() {
-                buf.push_str("{}");
-                return Ok(());
-            }
-
-            buf.push('{');
-            buf.push('\n');
-
             let keys = ordered_keys(map);
-            for (idx, key) in keys.iter().enumerate() {
+            write_delimited_entries(buf, indent, '{', '}', keys.len(), |idx, buf| {
+                let key = keys[idx];
                 let value = map
-                    .get(*key)
+                    .get(key)
                     .ok_or_else(|| internal_datafusion_err!("Missing key in object!"))?;
 
-                push_indent(buf, indent + 1);
-                buf.push_str(&json_key_to_string(key)?);
+                buf.push_str(&to_json_string!(key)?);
                 buf.push_str(": ");
-                write_ordered_json(value, buf, indent + 1)?;
-
-                if idx + 1 != keys.len() {
-                    buf.push(',');
-                }
-                buf.push('\n');
-            }
-
-            push_indent(buf, indent);
-            buf.push('}');
+                write_ordered_json(value, buf, indent + 1)
+            })?;
         }
     }
 

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

@@ -469,6 +469,23 @@ fn wrap_dictionary_scalar(key_type: &DataType, value: ScalarValue) -> ScalarValu
     ScalarValue::Dictionary(Box::new(key_type.clone()), Box::new(value))
 }
 
+fn update_extreme(
+    extreme: &mut ScalarValue,
+    values: &[ArrayRef],
+    batch_extreme: fn(&ArrayRef) -> Result<ScalarValue>,
+    is_max: bool,
+) -> Result<()> {
+    let delta = batch_extreme(&values[0])?;
+    let current = &*extreme;
+    let next: Result<ScalarValue, DataFusionError> = if is_max {
+        min_max!(current, &delta, max)
+    } else {
+        min_max!(current, &delta, min)
+    };
+    *extreme = next?;
+    Ok(())
+}
+
 /// An accumulator to compute the maximum value
 #[derive(Debug, Clone)]
 pub struct MaxAccumulator {
@@ -486,12 +503,7 @@ impl MaxAccumulator {
 
 impl Accumulator for MaxAccumulator {
     fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
-        let values = &values[0];
-        let delta = &max_batch(values)?;
-        let new_max: Result<ScalarValue, DataFusionError> =
-            min_max!(&self.max, delta, max);
-        self.max = new_max?;
-        Ok(())
+        update_extreme(&mut self.max, values, max_batch, true)
     }
 
     fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
@@ -531,12 +543,7 @@ impl Accumulator for MinAccumulator {
     }
 
     fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
-        let values = &values[0];
-        let delta = &min_batch(values)?;
-        let new_min: Result<ScalarValue, DataFusionError> =
-            min_max!(&self.min, delta, min);
-        self.min = new_min?;
-        Ok(())
+        update_extreme(&mut self.min, values, min_batch, false)
     }
 
     fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
@@ -806,13 +813,11 @@ pub fn min_batch(values: &ArrayRef) -> Result<ScalarValue> {
                 min_binary_view
             )
         }
-        DataType::Struct(_) => min_max_batch_generic(values, Ordering::Greater)?,
-        DataType::List(_) => min_max_batch_generic(values, Ordering::Greater)?,
-        DataType::LargeList(_) => min_max_batch_generic(values, Ordering::Greater)?,
-        DataType::FixedSizeList(_, _) => {
-            min_max_batch_generic(values, Ordering::Greater)?
-        }
-        DataType::Dictionary(_, _) => min_max_batch_generic(values, Ordering::Greater)?,
+        DataType::Struct(_)
+        | DataType::List(_)
+        | DataType::LargeList(_)
+        | DataType::FixedSizeList(_, _)
+        | DataType::Dictionary(_, _) => min_max_batch_generic(values, Ordering::Greater)?,
         _ => min_max_batch!(values, min),
     })
 }
@@ -868,11 +873,11 @@ pub fn max_batch(values: &ArrayRef) -> Result<ScalarValue> {
             let value = value.map(|e| e.to_vec());
             ScalarValue::FixedSizeBinary(*size, value)
         }
-        DataType::Struct(_) => min_max_batch_generic(values, Ordering::Less)?,
-        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(_, _) => min_max_batch_generic(values, Ordering::Less)?,
+        DataType::Struct(_)
+        | DataType::List(_)
+        | DataType::LargeList(_)
+        | DataType::FixedSizeList(_, _)
+        | DataType::Dictionary(_, _) => min_max_batch_generic(values, Ordering::Less)?,
         _ => min_max_batch!(values, max),
     })
 }