[Epic] Unify `AggregateFunction` Interface (remove built in list of `AggregateFunction` s), improve the system

[alamb]

Is your feature request related to a problem or challenge?

For many of the same reasons as listed on #8045, having two types of aggregate functions ("built in" -- datafusion::physical_plan::aggregates::AggregateFunction) and AggregateUDF is problematic for two reasons:

1. There are some features not available to User Defined Aggregate Functions (such as the faster GroupsAccumulator interface)
2. Users can not easily choose which aggregate functions to include (for example, do they want to allow (and pay the code size / compile time) for the Statistical and Approximate functions

The second also ends up causing pushback on adding new aggregates like ARRAY_SUM in #8325 and geospatial support #7859.

Describe the solution you'd like

I propose moving DataFusion to only use AggregateUDFs and remove the built in list of AggregateFunctions for the same reasons as #8045

We will keep the existing AggregateUDF interface as much as possible, while also potentially providing an easier way to define them.

New AggregateUDF is in functions-aggregate crate
Old Aggregate functions are in datafusion/physical-expr/src/aggregate

Describe alternatives you've considered

Additional context

Proposed implementation steps:

• Use lowercase name for aggregate function #10695
• Implement trait based API for defining AggregateUDF #8710
• Support GroupsAccumulator accumulator for User Defined Functions #8793
• Support ORDER BY in AggregateUDF  #8984
• Move AggregateExpr, PhysicalExpr and PhysicalSortExpr to physical-expr-core #9926
• Simplify API Add a AggregateUDFImpl::simplfy() API #9526
• Covariance Sample Move Covariance (Sample) covar / covar_samp to be a User Defined Aggregate Function #10372
• Covariance Population Move Covariance (Population) covar_pop to be a User Defined Aggregate Function #10418
• Count Move Count to functions-aggregate, update MSRV to rust 1.75 #10484
• Median Move Median to functions-aggregate and Introduce Numeric signature #10644
• First and Last Make FirstValue an UDAF, Change AggregateUDFImpl::accumulator signature, support ORDER BY for UDAFs #9874 Convert first, last aggregate function to UDAF #10648
• Sum Introduce Sum UDAF #10651
• Expression API Introduce expr builder for aggregate function #10560
• Variance Sample Convert Variance Sample to UDAF #10667
• Variance Population Convert Variance Population to UDAF #10668
• Stddev Convert stddev to udaf #10827
• Approx Distinct Convert approx_distinct to UDAF #10851
• approx_median Convert approx_median to UDAF #10838
• approx_percentile_cont approx_percentile_cont_with_weight Convert ApproxPercentileCont and `ApproxPercentileContWithWeight to UDAF #10870
• regr Convert Regr to UDAF #10883
• correlation Convert Correlation to UDAF #10884
• BitAnd / Or / Xor Convert BitAnd, BitOr, BitXor to UDAF #10907
• Average Convert Average to UDAF #10942
• Min Max Convert Min and Max to UDAF  #10943
• String agg Convert StringAgg to UDAF #10945
• ArrayAgg Convert ArrayAgg to UDAF #10999
• BoolAndOR Convert BoolAndOr to UDAF #11008
• Convert Min and Max to UDAF  #10943
• Remove special casting of Min / Max built in AggregateFunctions #11151

Move rust test to sqllogictest if possible #10384

Good first issue list

• grouping Convert Grouping to UDAF #10906

Pending

• nth_value
• array_agg_distinct
• array_agg_ordered

impl FromStr for AggregateFunction {
    type Err = DataFusionError;
    fn from_str(name: &str) -> Result<AggregateFunction> {
        Ok(match name {
            // general
            "avg" => AggregateFunction::Avg,
            "bit_and" => AggregateFunction::BitAnd,
            "bit_or" => AggregateFunction::BitOr,
            "bit_xor" => AggregateFunction::BitXor,
            "bool_and" => AggregateFunction::BoolAnd,
            "bool_or" => AggregateFunction::BoolOr,
            "max" => AggregateFunction::Max,
            "mean" => AggregateFunction::Avg,
            "min" => AggregateFunction::Min,
            "array_agg" => AggregateFunction::ArrayAgg,
            "nth_value" => AggregateFunction::NthValue,
            "string_agg" => AggregateFunction::StringAgg,
            // statistical
            "corr" => AggregateFunction::Correlation,
            // other
            "grouping" => AggregateFunction::Grouping,
            _ => {
                return plan_err!("There is no built-in function named {name}");
            }
        })
    }
}

Feel free to file an issue if you are interested in working on any of the above in the pending list.

[jayzhan211]

I will work on FirstValue UDF together with #9249

[jayzhan211]

Can we introduce state_fields and fields for AggregateUDFImpl. We can see that types in AggregateUDFImpl are for building fields, why not just return fields directly, we can not only define the types but also the field name and is_nullalbe.

https://github.com/apache/arrow-datafusion/blob/e337832946fc69f6ceccd14b96a071cc2cd4693d/datafusion/physical-plan/src/udaf.rs#L86-L106

[alamb]

Thank you @jayzhan211 -- I am traveling this week so I am very behind on reviews. I will try and respond later this week

[jayzhan211]

Thank you @jayzhan211 -- I am traveling this week so I am very behind on reviews. I will try and respond later this week

I leave the comment so I won't forget.
Have a good trip 😊

[jayzhan211]

I have another question. I think our goal for the aggregate function is similar to functions. we will move them into separate crate. If we need to avoid importing physical-expr. It seems we need to move some struct from physical-expr to expr, like PhysicalExpr and pub type LexOrdering = Vec<PhysicalSortExpr>;

https://github.com/apache/arrow-datafusion/blob/3eeb108125b35424baac39dd20ba88433b347419/datafusion/physical-expr/src/aggregate/first_last.rs#L44-L45

But, does moving PhysicalExpr to datafusion_expr make sense? 😕 Where/When should convert expr to physical-expr like sort_exprs to LexOrdering. Should we go through the whole process in the aggregate-functions crate like what functions do or we should separate logical-expr and physical-expr for aggregate-functions and find a way to link between them (convert from logical-expr to physical-expr)?