Aggregate UDF's hard ordering requirement is dropped when it has no reverse_expr

[killme2008]

Describe the bug

An aggregate with ORDER BY gets unsorted input if its UDAF keeps the default order_sensitivity() (HardRequirement) and has no reverse expression (default reverse_expr() is NotSupported). No SortExec is planned and the result depends on input order.

In get_finer_aggregate_exprs_requirement, when the input doesn't already satisfy the forward requirement, the requirement is only set inside if let Some(reverse_aggr_expr) = aggr_expr.reverse_expr() (L2996). With no reverse expression, nothing is recorded and the hard requirement is lost.

Built-in order-sensitive aggregates all implement reverse_expr, so they aren't affected. #17011 was this bug showing up in string_agg, fixed in #17165 by adding reverse_expr to StringAgg rather than in the planner.

To Reproduce

DataFusion 55.1.0. The UDAF below delegates to nth_value and keeps the default reverse_expr and order_sensitivity:

#[derive(Debug, PartialEq, Eq, Hash)]
struct NthValueNoReverse(Arc<AggregateUDF>);

impl AggregateUDFImpl for NthValueNoReverse {
    fn name(&self) -> &str {
        "nth_value_no_reverse"
    }
    fn signature(&self) -> &Signature {
        self.0.signature()
    }
    fn return_type(&self, arg_types: &[DataType]) -> Result<DataType> {
        self.0.return_type(arg_types)
    }
    fn accumulator(&self, acc_args: AccumulatorArgs) -> Result<Box<dyn Accumulator>> {
        self.0.accumulator(acc_args)
    }
    fn state_fields(&self, args: StateFieldsArgs) -> Result<Vec<FieldRef>> {
        self.0.state_fields(args)
    }
}

#[tokio::main]
async fn main() -> Result<()> {
    let ctx = SessionContext::new_with_config(SessionConfig::new().with_target_partitions(1));
    let schema = Arc::new(Schema::new(vec![
        Field::new("ts", DataType::Int64, false),
        Field::new("v", DataType::Float64, false),
    ]));
    let batch = RecordBatch::try_new(
        schema.clone(),
        vec![
            Arc::new(Int64Array::from(vec![3, 1, 2])),
            Arc::new(Float64Array::from(vec![30.0, 10.0, 20.0])),
        ],
    )?;
    ctx.register_table("t", Arc::new(MemTable::try_new(schema, vec![vec![batch]])?))?;
    ctx.register_udaf(AggregateUDF::new_from_impl(NthValueNoReverse(nth_value_udaf())));

    let sql = "SELECT nth_value_no_reverse(v, 1 ORDER BY ts) FROM t";
    ctx.sql(sql).await?.show().await?;
    ctx.sql(&format!("EXPLAIN {sql}")).await?.show().await?;
    Ok(())
}

It returns 30.0, and the plan has no sort:

AggregateExec: mode=Single, gby=[], aggr=[nth_value_no_reverse(t.v,Int64(1)) ORDER BY [t.ts ASC NULLS LAST]]
  DataSourceExec: partitions=1, partition_sizes=[1]

The built-in nth_value(v, 1 ORDER BY ts) returns 10.0 with a SortExec: expr=[ts@0 ASC NULLS LAST] under the aggregate. Putting both in one query makes the custom one correct too, since the built-in's requirement sorts the shared input.

Expected behavior

Without a reverse expression, the forward requirement should be adopted, or reported as a conflict with another hard requirement, same as the reverse branch already does:

} else if forward_finer.is_some() {
    requirement = Some(aggr_req);
} else if !include_soft_requirement {
    return not_impl_err!("Conflicting ordering requirements in aggregate functions is not supported");
}

The catch: ordered-set aggregates rely on the current behavior

With only that change, aggregate.slt fails: percentile_cont(0.5) WITHIN GROUP (ORDER BY x) now gets a sort on x:

-   03)----SortExec: TopK(fetch=10000), expr=[g@0 ASC NULLS LAST], preserve_partitioning=[false]
+   03)----SortExec: TopK(fetch=10000), expr=[g@0 ASC NULLS LAST, x@1 ASC NULLS LAST], preserve_partitioning=[false]

percentile_cont, approx_percentile_cont and approx_percentile_cont_with_weight store WITHIN GROUP (ORDER BY x) as the aggregate's ORDER BY, keep the default HardRequirement, and have no reverse expression. They don't need sorted input, so today they work only because this requirement gets dropped. None of the existing sensitivities fits them:

• Insensitive: AggregateExprBuilder::build drops order_bys for insensitive aggregates (aggregate.rs#L270), but these functions read the sort direction from acc_args.order_bys (percentile_cont.rs#L279). WITHIN GROUP (ORDER BY x DESC) would be computed as ascending.
• Beneficial: avoids the sort and keeps order_bys, but it says sorted input makes the aggregate cheaper, which isn't true here, and each function would have to implement with_beneficial_ordering or fail at udaf.rs#L702.

So fixing the planner needs a way to say "the ORDER BY is a parameter, not an input ordering" first, e.g. a new sensitivity, or having the planner skip aggregates with supports_within_group_clause(). Which way would you prefer? I'm happy to open a PR once that's settled.

Additional context

Found in GreptimeDB, which wraps aggregates into state/merge UDAFs for distributed execution. The wrappers can't expose the inner reverse expression because the partial and final stages are planned separately, so nth_value(... ORDER BY ...) ran on unsorted input.