SQL unparser drops ORDER BY inside ordered aggregates (last_value, first_value, array_agg, string_agg)

[harrydevforlife]

Describe the bug

The SQL unparser drops the ORDER BY inside ordered aggregate functions (last_value, first_value, array_agg, string_agg, …). The unparsed SQL is still valid, so it runs without an error, but it computes a different result. Any tool that sends unparsed SQL to another engine returns wrong answers without any warning. datafusion-federation is one example.

In datafusion/sql/src/unparser/expr.rs, the Expr::AggregateFunction branch only uses order_by for WITHIN GROUP aggregates. For every other aggregate it builds FunctionArgumentList { clauses: vec![], .. }, so the ordering never reaches the AST. This is still the case on main (checked at 5a09d99) and in 55.1.0.

This is the same kind of loss as #25462 (IGNORE NULLS / RESPECT NULLS, fix in #25475), in the same code.

To Reproduce

use datafusion::{error::Result, prelude::*, sql::unparser::plan_to_sql};

#[tokio::main]
async fn main() -> Result<()> {
    let ctx = SessionContext::new();
    ctx.sql("CREATE TABLE t (a INT, b INT, g INT) AS VALUES (1, 2, 1), (3, 1, 1)").await?;
    for sql in [
        "SELECT g, last_value(a ORDER BY b) FROM t GROUP BY g",
        "SELECT g, first_value(a ORDER BY b DESC) FROM t GROUP BY g",
        "SELECT g, array_agg(a ORDER BY b) FROM t GROUP BY g",
        "SELECT g, string_agg(CAST(a AS VARCHAR), ',' ORDER BY b) FROM t GROUP BY g",
    ] {
        let plan = ctx.sql(sql).await?.into_unoptimized_plan();
        println!("{}", plan_to_sql(&plan)?);
    }
    Ok(())
}

Output (DataFusion 55.1.0):

SELECT t.g, last_value(t.a) FROM t GROUP BY t.g
SELECT t.g, first_value(t.a) FROM t GROUP BY t.g
SELECT t.g, array_agg(t.a) FROM t GROUP BY t.g
SELECT t.g, string_agg(CAST(t.a AS VARCHAR), ',') FROM t GROUP BY t.g

Running the unparsed SQL gives a different result from the original for all four queries.

Expected behavior

The ordering is kept, for example SELECT t.g, last_value(t.a ORDER BY t.b ASC NULLS LAST) FROM t GROUP BY t.g, so that parsing, planning and unparsing a query gives back an equivalent query.

Additional context

A possible fix: when order_by is non-empty and the function doesn't support WITHIN GROUP, emit it as an argument clause:

clauses: if order_by.is_empty() || agg.func.supports_within_group_clause() {
    vec![]
} else {
    vec![ast::FunctionArgumentClause::OrderBy(
        order_by.iter().map(|s| self.sort_to_sql(s)).collect::<Result<Vec<_>>>()?,
    )]
},

It could be tested with a round-trip case in datafusion/sql/tests/cases/plan_to_sql.rs for each of the four functions above. Found while federating last_value(v ORDER BY t) to remote engines with datafusion-federation 0.5.7.

[Vanzeren]

take