fix: render aggregate labels without changing EXPLAIN

Aggregate labels came from `UdafHumanDisplayBuilder`. Rendering its
FILTER and ORDER BY clauses in SQL syntax for labels also changed
physical EXPLAIN text with the option off and hid literal types such
as `Float32(1)`. It also broke `reverse_expr`: it still looks for
`ORDER BY [`, so a reversed `first_value` showed as `last_value` with
the old sort direction.

Restore the builder and render aggregate labels in the labeling step,
as window labels already are. Ordered-set aggregates now get labels
such as `percentile_cont(0.5) WITHIN GROUP (ORDER BY c)`.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: haochunchang

datafusion/core/src/physical_planner.rs

@@ -4909,7 +4909,7 @@ mod tests {
                 vec![expr],
             )
             .await?,
-            "AggregateExec: mode=Single, gby=[], aggr=[sum(?table?.column1) FILTER (WHERE ?table?.column2 <= 0) as agg]"
+            "AggregateExec: mode=Single, gby=[], aggr=[sum(?table?.column1) FILTER (WHERE ?table?.column2 <= Int64(0)) as agg]"
         );
 
         Ok(())
@@ -4933,7 +4933,7 @@ mod tests {
                 vec![expr],
             )
             .await?,
-            "AggregateExec: mode=Single, gby=[], aggr=[first_value(?table?.column1) RESPECT NULLS ORDER BY ?table?.column2 NULLS FIRST as agg]"
+            "AggregateExec: mode=Single, gby=[], aggr=[first_value(?table?.column1) RESPECT NULLS ORDER BY [?table?.column2 ASC NULLS FIRST] as agg]"
         );
 
         Ok(())

datafusion/core/tests/sql/column_labels.rs

@@ -155,6 +155,25 @@ async fn column_labels_aggregate_functions() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn column_labels_ordered_set_aggregates() -> Result<()> {
+    let ctx = labels_ctx().await?;
+    let queries = [
+        "SELECT percentile_cont(0.5) WITHIN GROUP (ORDER BY c) FROM t",
+        "SELECT percentile_cont(0.5) WITHIN GROUP (ORDER BY c DESC) FROM t",
+        "SELECT percentile_cont(c, 0.5) FROM t",
+    ];
+    assert_snapshot!(labels(&ctx, &queries).await?, @r"
+    SELECT percentile_cont(0.5) WITHIN GROUP (ORDER BY c) FROM t
+      percentile_cont(Float64(0.5)) WITHIN GROUP [t.c ASC NULLS LAST] => percentile_cont(0.5) WITHIN GROUP (ORDER BY c)
+    SELECT percentile_cont(0.5) WITHIN GROUP (ORDER BY c DESC) FROM t
+      percentile_cont(Float64(0.5)) WITHIN GROUP [t.c DESC NULLS FIRST] => percentile_cont(0.5) WITHIN GROUP (ORDER BY c DESC)
+    SELECT percentile_cont(c, 0.5) FROM t
+      percentile_cont(t.c,Float64(0.5)) => percentile_cont(c, 0.5)
+    ");
+    Ok(())
+}
+
 #[tokio::test]
 async fn column_labels_window_functions() -> Result<()> {
     let ctx = labels_ctx().await?;

datafusion/expr/src/udaf.rs

@@ -32,9 +32,9 @@ use datafusion_expr_common::operator::Operator;
 use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
 
 use crate::expr::{
-    AggregateFunction, AggregateFunctionParams, ExprListDisplay, SortListDisplay,
-    WindowFunctionParams, schema_name_from_exprs,
-    schema_name_from_exprs_comma_separated_without_space, schema_name_from_sorts,
+    AggregateFunction, AggregateFunctionParams, ExprListDisplay, WindowFunctionParams,
+    schema_name_from_exprs, schema_name_from_exprs_comma_separated_without_space,
+    schema_name_from_sorts,
 };
 use crate::function::{
     AccumulatorArgs, AggregateFunctionSimplification, StateFieldsArgs,
@@ -1131,23 +1131,20 @@ pub fn udaf_default_schema_name<F: AggregateUDFImpl + ?Sized>(
 ///
 /// # Example
 /// ```
-/// # use datafusion_expr::{col, lit};
+/// # use datafusion_expr::col;
 /// # use datafusion_expr::expr::AggregateFunctionParams;
 /// # use datafusion_expr::UdafHumanDisplayBuilder;
 /// # let params = AggregateFunctionParams {
 /// #     args: vec![col("x")],
 /// #     distinct: false,
-/// #     filter: Some(Box::new(col("y").gt(lit(5i64)))),
-/// #     order_by: vec![col("z").sort(false, true)],
+/// #     filter: Some(Box::new(col("y").gt(col("z")))),
+/// #     order_by: vec![],
 /// #     null_treatment: None,
 /// # };
 /// let human_display = UdafHumanDisplayBuilder::new("my_udaf", &params)
 ///     .build()
 ///     .unwrap();
-/// assert_eq!(
-///     human_display,
-///     "my_udaf(x) FILTER (WHERE y > 5) ORDER BY z DESC"
-/// );
+/// assert_eq!(human_display, "my_udaf(x) FILTER (WHERE y > z)");
 /// ```
 #[derive(Debug)]
 pub struct UdafHumanDisplayBuilder<'a> {
@@ -1188,13 +1185,14 @@ impl<'a> UdafHumanDisplayBuilder<'a> {
         }
 
         if let Some(filter) = filter {
-            schema_name
-                .write_fmt(format_args!(" FILTER (WHERE {})", filter.human_display()))?;
+            schema_name.write_fmt(format_args!(" FILTER (WHERE {filter})"))?;
         }
 
         if !order_by.is_empty() {
-            schema_name
-                .write_fmt(format_args!(" ORDER BY {}", SortListDisplay(order_by)))?;
+            schema_name.write_fmt(format_args!(
+                " ORDER BY [{}]",
+                schema_name_from_sorts(order_by)?
+            ))?;
         }
 
         Ok(schema_name)

datafusion/sql/src/column_labels.rs

@@ -33,7 +33,8 @@ use datafusion_common::metadata::{FieldMetadata, column_label_metadata};
 use datafusion_common::tree_node::{Transformed, TreeNode};
 use datafusion_common::{Column, DFSchema, Dependency, Result, TableReference};
 use datafusion_expr::expr::{
-    Alias, ExprListDisplay, SortListDisplay, WindowFunction, WindowFunctionParams,
+    AggregateFunction, AggregateFunctionParams, Alias, ExprListDisplay, SortListDisplay,
+    WindowFunction, WindowFunctionParams,
 };
 use datafusion_expr::utils::grouping_set_to_exprlist;
 use datafusion_expr::{
@@ -189,7 +190,7 @@ fn trace_column(plan: &LogicalPlan, idx: usize) -> Result<Option<Expr>> {
                 }
                 idx -= 1;
             }
-            resolve_expr(&aggregate.aggr_expr[idx], &aggregate.input)
+            resolve_aggregate_expr(&aggregate.aggr_expr[idx], &aggregate.input)
         }
         LogicalPlan::Window(window) => {
             let input_len = window.input.schema().fields().len();
@@ -311,6 +312,76 @@ fn is_count_star_window_alias(column: &Column, alias: &str) -> bool {
         && alias == column.name.replacen(PLANNED, "count(*)", 1)
 }
 
+/// Like [`resolve_expr`] for an expression of an `Aggregate` node. An
+/// aggregate function is rendered as `name(args)` followed by its
+/// `DISTINCT`, null treatment, `FILTER` and `ORDER BY` parts. An ordered-set
+/// aggregate is rendered with `WITHIN GROUP`, like in the query.
+fn resolve_aggregate_expr(expr: &Expr, input: &LogicalPlan) -> Result<Option<Expr>> {
+    let Expr::AggregateFunction(aggregate_function) = expr else {
+        return resolve_expr(expr, input);
+    };
+    let AggregateFunction { func, params } = aggregate_function;
+    let AggregateFunctionParams {
+        args,
+        distinct,
+        filter,
+        order_by,
+        null_treatment,
+    } = params;
+
+    let Some(args) = args
+        .iter()
+        .map(|e| resolve_expr(e, input))
+        .collect::<Result<Option<Vec<_>>>>()?
+    else {
+        return Ok(None);
+    };
+    let mut resolved_order_by = Vec::with_capacity(order_by.len());
+    for sort in order_by {
+        let Some(expr) = resolve_expr(&sort.expr, input)? else {
+            return Ok(None);
+        };
+        resolved_order_by.push(sort.with_expr(expr));
+    }
+    let order_by = resolved_order_by;
+    let filter = match filter {
+        Some(filter) => match resolve_expr(filter, input)? {
+            Some(filter) => Some(filter),
+            None => return Ok(None),
+        },
+        None => None,
+    };
+
+    let within_group = func.supports_within_group_clause() && !order_by.is_empty();
+    // The planner puts the ordering expression of a WITHIN GROUP clause first
+    // in the args, but the query only has the direct args there
+    let args = if within_group { &args[1..] } else { &args[..] };
+    let mut label = format!(
+        "{}({}{})",
+        func.name(),
+        if *distinct { "DISTINCT " } else { "" },
+        ExprListDisplay::comma_separated(args)
+    );
+    if within_group {
+        write!(
+            label,
+            " WITHIN GROUP (ORDER BY {})",
+            SortListDisplay(&order_by)
+        )?;
+    }
+    if let Some(null_treatment) = null_treatment {
+        write!(label, " {null_treatment}")?;
+    }
+    if let Some(filter) = filter {
+        write!(label, " FILTER (WHERE {})", filter.human_display())?;
+    }
+    if !within_group && !order_by.is_empty() {
+        write!(label, " ORDER BY {}", SortListDisplay(&order_by))?;
+    }
+
+    Ok(Some(Expr::Column(Column::new_unqualified(label))))
+}
+
 /// Like [`resolve_expr`] for an expression of a `Window` node. A window
 /// function is rendered as `name(args) OVER (...)` without the default parts
 /// of its window definition.

datafusion/sqllogictest/test_files/agg_func_substitute.slt

@@ -79,9 +79,9 @@ logical_plan
 03)----TableScan: multiple_ordered_table projection=[a, c]
 physical_plan
 01)ProjectionExec: expr=[a@0 as a, nth_value(multiple_ordered_table.c,Int64(1) + Int64(100)) ORDER BY [multiple_ordered_table.c ASC NULLS LAST]@1 as result]
-02)--AggregateExec: mode=FinalPartitioned, gby=[a@0 as a], aggr=[nth_value(multiple_ordered_table.c, 101) ORDER BY multiple_ordered_table.c as nth_value(multiple_ordered_table.c,Int64(1) + Int64(100)) ORDER BY [multiple_ordered_table.c ASC NULLS LAST]], ordering_mode=Sorted
+02)--AggregateExec: mode=FinalPartitioned, gby=[a@0 as a], aggr=[nth_value(multiple_ordered_table.c, 101) ORDER BY [multiple_ordered_table.c ASC NULLS LAST] as nth_value(multiple_ordered_table.c,Int64(1) + Int64(100)) ORDER BY [multiple_ordered_table.c ASC NULLS LAST]], ordering_mode=Sorted
 03)----RepartitionExec: partitioning=Hash([a@0], 4), input_partitions=4, preserve_order=true, sort_exprs=a@0 ASC NULLS LAST
-04)------AggregateExec: mode=Partial, gby=[a@0 as a], aggr=[nth_value(multiple_ordered_table.c, 101) ORDER BY multiple_ordered_table.c as nth_value(multiple_ordered_table.c,Int64(1) + Int64(100)) ORDER BY [multiple_ordered_table.c ASC NULLS LAST]], ordering_mode=Sorted
+04)------AggregateExec: mode=Partial, gby=[a@0 as a], aggr=[nth_value(multiple_ordered_table.c, 101) ORDER BY [multiple_ordered_table.c ASC NULLS LAST] as nth_value(multiple_ordered_table.c,Int64(1) + Int64(100)) ORDER BY [multiple_ordered_table.c ASC NULLS LAST]], ordering_mode=Sorted
 05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, maintains_sort_order=true
 06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, c], output_orderings=[[a@0 ASC NULLS LAST], [c@1 ASC NULLS LAST]], file_type=csv, has_header=true
 

datafusion/sqllogictest/test_files/aggregates_simplify.slt

@@ -598,9 +598,9 @@ logical_plan
 01)Aggregate: groupBy=[[distinct_simplify_t.g]], aggr=[[min(distinct_simplify_t.v) FILTER (WHERE distinct_simplify_t.v > Int32(3)) AS min(DISTINCT distinct_simplify_t.v) FILTER (WHERE distinct_simplify_t.v > Int64(3))]]
 02)--TableScan: distinct_simplify_t projection=[g, v]
 physical_plan
-01)AggregateExec: mode=FinalPartitioned, gby=[g@0 as g], aggr=[min(distinct_simplify_t.v) FILTER (WHERE distinct_simplify_t.v > 3) as min(DISTINCT distinct_simplify_t.v) FILTER (WHERE distinct_simplify_t.v > Int64(3))]
+01)AggregateExec: mode=FinalPartitioned, gby=[g@0 as g], aggr=[min(distinct_simplify_t.v) FILTER (WHERE distinct_simplify_t.v > Int32(3)) as min(DISTINCT distinct_simplify_t.v) FILTER (WHERE distinct_simplify_t.v > Int64(3))]
 02)--RepartitionExec: partitioning=Hash([g@0], 4), input_partitions=1
-03)----AggregateExec: mode=Partial, gby=[g@0 as g], aggr=[min(distinct_simplify_t.v) FILTER (WHERE distinct_simplify_t.v > 3) as min(DISTINCT distinct_simplify_t.v) FILTER (WHERE distinct_simplify_t.v > Int64(3))]
+03)----AggregateExec: mode=Partial, gby=[g@0 as g], aggr=[min(distinct_simplify_t.v) FILTER (WHERE distinct_simplify_t.v > Int32(3)) as min(DISTINCT distinct_simplify_t.v) FILTER (WHERE distinct_simplify_t.v > Int64(3))]
 04)------DataSourceExec: partitions=1, partition_sizes=[1]
 
 # min/max(DISTINCT) over floats now sees the raw values. Before this rule, the

datafusion/sqllogictest/test_files/expr.slt

@@ -2606,7 +2606,7 @@ logical_plan
 03)----TableScan: t projection=[a]
 physical_plan
 01)ProjectionExec: expr=[min(t.a) FILTER (WHERE t.a > Int64(1))@0 as x]
-02)--AggregateExec: mode=Single, gby=[], aggr=[min(t.a) FILTER (WHERE t.a > 1) as min(t.a) FILTER (WHERE t.a > Int64(1))]
+02)--AggregateExec: mode=Single, gby=[], aggr=[min(t.a) FILTER (WHERE t.a > Float32(1)) as min(t.a) FILTER (WHERE t.a > Int64(1))]
 03)----DataSourceExec: partitions=1, partition_sizes=[1]