killme2008 opened a new issue, #25676: URL: https://github.com/apache/datafusion/issues/25676
### 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`](https://github.com/apache/datafusion/blob/95bb0a0dfa48ca20d017faa55b2fc96ee2f300fe/datafusion/physical-plan/src/aggregates/mod.rs#L2959), 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](https://github.com/apache/datafusion/blob/95bb0a0dfa48ca20d017faa55b2fc96ee2f300fe/datafusion/physical-plan/src/aggregates/mod.rs#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`: ```rust #[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: ```rust } 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](https://github.com/apache/datafusion/blob/95bb0a0dfa48ca20d017faa55b2fc96ee2f300fe/datafusion/physical-expr/src/aggregate.rs#L270)), but these functions read the sort direction from `acc_args.order_bys` ([percentile_cont.rs#L279](https://github.com/apache/datafusion/blob/95bb0a0dfa48ca20d017faa55b2fc96ee2f300fe/datafusion/functions-aggregate/src/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](https://github.com/apache/datafusion/blob/95bb0a0dfa48ca20d017faa55b2fc96ee2f300fe/datafusion/expr/src/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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
