## Coverage Report

#### Created: 2025-11-22 06:00

next uncovered line (L), next uncovered region (R), next uncovered branch (B)

    /Users/andrewlamb/Software/datafusion/datafusion/expr/src/udaf.rs

    Line
    Count
    Source
    1
    // Licensed to the Apache Software Foundation (ASF) under one
    2
    // or more contributor license agreements.  See the NOTICE file
    3
    // distributed with this work for additional information
    4
    // regarding copyright ownership.  The ASF licenses this file
    5
    // to you under the Apache License, Version 2.0 (the
    6
    // "License"); you may not use this file except in compliance
    7
    // with the License.  You may obtain a copy of the License at
    8
    //
    9
    //   http://www.apache.org/licenses/LICENSE-2.0
    10
    //
    11
    // Unless required by applicable law or agreed to in writing,
    12
    // software distributed under the License is distributed on an
    13
    // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
    14
    // KIND, either express or implied.  See the License for the
    15
    // specific language governing permissions and limitations
    16
    // under the License.
    17
    18
    //! [`AggregateUDF`]: User Defined Aggregate Functions
    19
    20
    use std::any::Any;
    21
    use std::cmp::Ordering;
    22
    use std::fmt::{self, Debug, Formatter, Write};
    23
    use std::hash::{Hash, Hasher};
    24
    use std::sync::Arc;
    25
    use std::vec;
    26
    27
    use arrow::datatypes::{DataType, Field, FieldRef};
    28
    29
    use datafusion_common::{exec_err, not_impl_err, Result, ScalarValue, Statistics};
    30
    use datafusion_expr_common::dyn_eq::{DynEq, DynHash};
    31
    use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
    32
    33
    use crate::expr::{
    34
        schema_name_from_exprs, schema_name_from_exprs_comma_separated_without_space,
    35
        schema_name_from_sorts, AggregateFunction, AggregateFunctionParams, ExprListDisplay,
    36
        WindowFunctionParams,
    37
    };
    38
    use crate::function::{
    39
        AccumulatorArgs, AggregateFunctionSimplification, StateFieldsArgs,
    40
    };
    41
    use crate::groups_accumulator::GroupsAccumulator;
    42
    use crate::udf_eq::UdfEq;
    43
    use crate::utils::format_state_name;
    44
    use crate::utils::AggregateOrderSensitivity;
    45
    use crate::{expr_vec_fmt, Accumulator, Expr};
    46
    use crate::{Documentation, Signature};
    47
    48
    /// Logical representation of a user-defined [aggregate function] (UDAF).
    49
    ///
    50
    /// An aggregate function combines the values from multiple input rows
    51
    /// into a single output "aggregate" (summary) row. It is different
    52
    /// from a scalar function because it is stateful across batches. User
    53
    /// defined aggregate functions can be used as normal SQL aggregate
    54
    /// functions (`GROUP BY` clause) as well as window functions (`OVER`
    55
    /// clause).
    56
    ///
    57
    /// `AggregateUDF` provides DataFusion the information needed to plan and call
    58
    /// aggregate functions, including name, type information, and a factory
    59
    /// function to create an [`Accumulator`] instance, to perform the actual
    60
    /// aggregation.
    61
    ///
    62
    /// For more information, please see [the examples]:
    63
    ///
    64
    /// 1. For simple use cases, use [`create_udaf`] (examples in [`simple_udaf.rs`]).
    65
    ///
    66
    /// 2. For advanced use cases, use [`AggregateUDFImpl`] which provides full API
    67
    ///    access (examples in [`advanced_udaf.rs`]).
    68
    ///
    69
    /// # API Note
    70
    /// This is a separate struct from `AggregateUDFImpl` to maintain backwards
    71
    /// compatibility with the older API.
    72
    ///
    73
    /// [the examples]: https://github.com/apache/datafusion/tree/main/datafusion-examples#single-process
    74
    /// [aggregate function]: https://en.wikipedia.org/wiki/Aggregate_function
    75
    /// [`Accumulator`]: Accumulator
    76
    /// [`create_udaf`]: crate::expr_fn::create_udaf
    77
    /// [`simple_udaf.rs`]: https://github.com/apache/datafusion/blob/main/datafusion-examples/examples/simple_udaf.rs
    78
    /// [`advanced_udaf.rs`]: https://github.com/apache/datafusion/blob/main/datafusion-examples/examples/advanced_udaf.rs
    79
    #[derive(Debug, Clone, PartialOrd)]
    80
    pub struct AggregateUDF {
    81
        inner: Arc<dyn AggregateUDFImpl>,
    82
    }
    83
    84
    impl PartialEq for AggregateUDF {
    85
    0
        fn eq(&self, other: &Self) -> bool {
    86
    0
            self.inner.dyn_eq(other.inner.as_any())
    87
    0
        }
    88
    }
    89
    90
    impl Eq for AggregateUDF {}
    91
    92
    impl Hash for AggregateUDF {
    93
    0
        fn hash<H: Hasher>(&self, state: &mut H) {
    94
    0
            self.inner.dyn_hash(state)
    95
    0
        }
    96
    }
    97
    98
    impl fmt::Display for AggregateUDF {
    99
    0
        fn fmt(&self, f: &mut Formatter) -> fmt::Result {
    100
    0
            write!(f, "{}", self.name())
    101
    0
        }
    102
    }
    103
    104
    /// Arguments passed to [`AggregateUDFImpl::value_from_stats`]
    105
    #[derive(Debug)]
    106
    pub struct StatisticsArgs<'a> {
    107
        /// The statistics of the aggregate input
    108
        pub statistics: &'a Statistics,
    109
        /// The resolved return type of the aggregate function
    110
        pub return_type: &'a DataType,
    111
        /// Whether the aggregate function is distinct.
    112
        ///
    113
        /// ```sql
    114
        /// SELECT COUNT(DISTINCT column1) FROM t;
    115
        /// ```
    116
        pub is_distinct: bool,
    117
        /// The physical expression of arguments the aggregate function takes.
    118
        pub exprs: &'a [Arc<dyn PhysicalExpr>],
    119
    }
    120
    121
    impl AggregateUDF {
    122
        /// Create a new `AggregateUDF` from a `[AggregateUDFImpl]` trait object
    123
        ///
    124
        /// Note this is the same as using the `From` impl (`AggregateUDF::from`)
    125
    0
        pub fn new_from_impl<F>(fun: F) -> AggregateUDF
    126
    0
        where
    127
    0
            F: AggregateUDFImpl + 'static,
    128
        {
    129
    0
            Self::new_from_shared_impl(Arc::new(fun))
    130
    0
        }
    131
    132
        /// Create a new `AggregateUDF` from a `[AggregateUDFImpl]` trait object
    133
    0
        pub fn new_from_shared_impl(fun: Arc<dyn AggregateUDFImpl>) -> AggregateUDF {
    134
    0
            Self { inner: fun }
    135
    0
        }
    136
    137
        /// Return the underlying [`AggregateUDFImpl`] trait object for this function
    138
    0
        pub fn inner(&self) -> &Arc<dyn AggregateUDFImpl> {
    139
    0
            &self.inner
    140
    0
        }
    141
    142
        /// Adds additional names that can be used to invoke this function, in
    143
        /// addition to `name`
    144
        ///
    145
        /// If you implement [`AggregateUDFImpl`] directly you should return aliases directly.
    146
    0
        pub fn with_aliases(self, aliases: impl IntoIterator<Item = &'static str>) -> Self {
    147
    0
            Self::new_from_impl(AliasedAggregateUDFImpl::new(
    148
    0
                Arc::clone(&self.inner),
    149
    0
                aliases,
    150
            ))
    151
    0
        }
    152
    153
        /// Creates an [`Expr`] that calls the aggregate function.
    154
        ///
    155
        /// This utility allows using the UDAF without requiring access to
    156
        /// the registry, such as with the DataFrame API.
    157
    0
        pub fn call(&self, args: Vec<Expr>) -> Expr {
    158
    0
            Expr::AggregateFunction(AggregateFunction::new_udf(
    159
    0
                Arc::new(self.clone()),
    160
    0
                args,
    161
    0
                false,
    162
    0
                None,
    163
    0
                vec![],
    164
    0
                None,
    165
    0
            ))
    166
    0
        }
    167
    168
        /// Returns this function's name
    169
        ///
    170
        /// See [`AggregateUDFImpl::name`] for more details.
    171
    0
        pub fn name(&self) -> &str {
    172
    0
            self.inner.name()
    173
    0
        }
    174
    175
        /// Returns the aliases for this function.
    176
    0
        pub fn aliases(&self) -> &[String] {
    177
    0
            self.inner.aliases()
    178
    0
        }
    179
    180
        /// See [`AggregateUDFImpl::schema_name`] for more details.
    181
    0
        pub fn schema_name(&self, params: &AggregateFunctionParams) -> Result<String> {
    182
    0
            self.inner.schema_name(params)
    183
    0
        }
    184
    185
        /// Returns a human readable expression.
    186
        ///
    187
        /// See [`Expr::human_display`] for details.
    188
    0
        pub fn human_display(&self, params: &AggregateFunctionParams) -> Result<String> {
    189
    0
            self.inner.human_display(params)
    190
    0
        }
    191
    192
    0
        pub fn window_function_schema_name(
    193
    0
            &self,
    194
    0
            params: &WindowFunctionParams,
    195
    0
        ) -> Result<String> {
    196
    0
            self.inner.window_function_schema_name(params)
    197
    0
        }
    198
    199
        /// See [`AggregateUDFImpl::display_name`] for more details.
    200
    0
        pub fn display_name(&self, params: &AggregateFunctionParams) -> Result<String> {
    201
    0
            self.inner.display_name(params)
    202
    0
        }
    203
    204
    0
        pub fn window_function_display_name(
    205
    0
            &self,
    206
    0
            params: &WindowFunctionParams,
    207
    0
        ) -> Result<String> {
    208
    0
            self.inner.window_function_display_name(params)
    209
    0
        }
    210
    211
    0
        pub fn is_nullable(&self) -> bool {
    212
    0
            self.inner.is_nullable()
    213
    0
        }
    214
    215
        /// Returns this function's signature (what input types are accepted)
    216
        ///
    217
        /// See [`AggregateUDFImpl::signature`] for more details.
    218
    0
        pub fn signature(&self) -> &Signature {
    219
    0
            self.inner.signature()
    220
    0
        }
    221
    222
        /// Return the type of the function given its input types
    223
        ///
    224
        /// See [`AggregateUDFImpl::return_type`] for more details.
    225
    0
        pub fn return_type(&self, args: &[DataType]) -> Result<DataType> {
    226
    0
            self.inner.return_type(args)
    227
    0
        }
    228
    229
        /// Return the field of the function given its input fields
    230
        ///
    231
        /// See [`AggregateUDFImpl::return_field`] for more details.
    232
    0
        pub fn return_field(&self, args: &[FieldRef]) -> Result<FieldRef> {
    233
    0
            self.inner.return_field(args)
    234
    0
        }
    235
    236
        /// Return an accumulator the given aggregate, given its return datatype
    237
    0
        pub fn accumulator(&self, acc_args: AccumulatorArgs) -> Result<Box<dyn Accumulator>> {
    238
    0
            self.inner.accumulator(acc_args)
    239
    0
        }
    240
    241
        /// Return the fields used to store the intermediate state for this aggregator, given
    242
        /// the name of the aggregate, value type and ordering fields. See [`AggregateUDFImpl::state_fields`]
    243
        /// for more details.
    244
        ///
    245
        /// This is used to support multi-phase aggregations
    246
    0
        pub fn state_fields(&self, args: StateFieldsArgs) -> Result<Vec<FieldRef>> {
    247
    0
            self.inner.state_fields(args)
    248
    0
        }
    249
    250
        /// See [`AggregateUDFImpl::groups_accumulator_supported`] for more details.
    251
    0
        pub fn groups_accumulator_supported(&self, args: AccumulatorArgs) -> bool {
    252
    0
            self.inner.groups_accumulator_supported(args)
    253
    0
        }
    254
    255
        /// See [`AggregateUDFImpl::create_groups_accumulator`] for more details.
    256
    0
        pub fn create_groups_accumulator(
    257
    0
            &self,
    258
    0
            args: AccumulatorArgs,
    259
    0
        ) -> Result<Box<dyn GroupsAccumulator>> {
    260
    0
            self.inner.create_groups_accumulator(args)
    261
    0
        }
    262
    263
    0
        pub fn create_sliding_accumulator(
    264
    0
            &self,
    265
    0
            args: AccumulatorArgs,
    266
    0
        ) -> Result<Box<dyn Accumulator>> {
    267
    0
            self.inner.create_sliding_accumulator(args)
    268
    0
        }
    269
    270
    0
        pub fn coerce_types(&self, arg_types: &[DataType]) -> Result<Vec<DataType>> {
    271
    0
            self.inner.coerce_types(arg_types)
    272
    0
        }
    273
    274
        /// See [`AggregateUDFImpl::with_beneficial_ordering`] for more details.
    275
    0
        pub fn with_beneficial_ordering(
    276
    0
            self,
    277
    0
            beneficial_ordering: bool,
    278
    0
        ) -> Result<Option<AggregateUDF>> {
    279
    0
            self.inner
    280
    0
                .with_beneficial_ordering(beneficial_ordering)
    281
    0
                .map(|updated_udf| updated_udf.map(|udf| Self { inner: udf }))
    282
    0
        }
    283
    284
        /// Gets the order sensitivity of the UDF. See [`AggregateOrderSensitivity`]
    285
        /// for possible options.
    286
    0
        pub fn order_sensitivity(&self) -> AggregateOrderSensitivity {
    287
    0
            self.inner.order_sensitivity()
    288
    0
        }
    289
    290
        /// Reserves the `AggregateUDF` (e.g. returns the `AggregateUDF` that will
    291
        /// generate same result with this `AggregateUDF` when iterated in reverse
    292
        /// order, and `None` if there is no such `AggregateUDF`).
    293
    0
        pub fn reverse_udf(&self) -> ReversedUDAF {
    294
    0
            self.inner.reverse_expr()
    295
    0
        }
    296
    297
        /// Do the function rewrite
    298
        ///
    299
        /// See [`AggregateUDFImpl::simplify`] for more details.
    300
    0
        pub fn simplify(&self) -> Option<AggregateFunctionSimplification> {
    301
    0
            self.inner.simplify()
    302
    0
        }
    303
    304
        /// Returns true if the function is max, false if the function is min
    305
        /// None in all other cases, used in certain optimizations for
    306
        /// or aggregate
    307
    0
        pub fn is_descending(&self) -> Option<bool> {
    308
    0
            self.inner.is_descending()
    309
    0
        }
    310
    311
        /// Return the value of this aggregate function if it can be determined
    312
        /// entirely from statistics and arguments.
    313
        ///
    314
        /// See [`AggregateUDFImpl::value_from_stats`] for more details.
    315
    0
        pub fn value_from_stats(
    316
    0
            &self,
    317
    0
            statistics_args: &StatisticsArgs,
    318
    0
        ) -> Option<ScalarValue> {
    319
    0
            self.inner.value_from_stats(statistics_args)
    320
    0
        }
    321
    322
        /// See [`AggregateUDFImpl::default_value`] for more details.
    323
    0
        pub fn default_value(&self, data_type: &DataType) -> Result<ScalarValue> {
    324
    0
            self.inner.default_value(data_type)
    325
    0
        }
    326
    327
        /// See [`AggregateUDFImpl::supports_null_handling_clause`] for more details.
    328
    0
        pub fn supports_null_handling_clause(&self) -> bool {
    329
    0
            self.inner.supports_null_handling_clause()
    330
    0
        }
    331
    332
        /// See [`AggregateUDFImpl::supports_within_group_clause`] for more details.
    333
    0
        pub fn supports_within_group_clause(&self) -> bool {
    334
    0
            self.inner.supports_within_group_clause()
    335
    0
        }
    336
    337
        /// Returns the documentation for this Aggregate UDF.
    338
        ///
    339
        /// Documentation can be accessed programmatically as well as
    340
        /// generating publicly facing documentation.
    341
    0
        pub fn documentation(&self) -> Option<&Documentation> {
    342
    0
            self.inner.documentation()
    343
    0
        }
    344
    }
    345
    346
    impl<F> From<F> for AggregateUDF
    347
    where
    348
        F: AggregateUDFImpl + Send + Sync + 'static,
    349
    {
    350
    0
        fn from(fun: F) -> Self {
    351
    0
            Self::new_from_impl(fun)
    352
    0
        }
    353
    }
    354
    355
    /// Trait for implementing [`AggregateUDF`].
    356
    ///
    357
    /// This trait exposes the full API for implementing user defined aggregate functions and
    358
    /// can be used to implement any function.
    359
    ///
    360
    /// See [`advanced_udaf.rs`] for a full example with complete implementation and
    361
    /// [`AggregateUDF`] for other available options.
    362
    ///
    363
    /// [`advanced_udaf.rs`]: https://github.com/apache/datafusion/blob/main/datafusion-examples/examples/advanced_udaf.rs
    364
    ///
    365
    /// # Basic Example
    366
    /// ```
    367
    /// # use std::any::Any;
    368
    /// # use std::sync::{Arc, LazyLock};
    369
    /// # use arrow::datatypes::{DataType, FieldRef};
    370
    /// # use datafusion_common::{DataFusionError, plan_err, Result};
    371
    /// # use datafusion_expr::{col, ColumnarValue, Signature, Volatility, Expr, Documentation};
    372
    /// # use datafusion_expr::{AggregateUDFImpl, AggregateUDF, Accumulator, function::{AccumulatorArgs, StateFieldsArgs}};
    373
    /// # use datafusion_expr::window_doc_sections::DOC_SECTION_AGGREGATE;
    374
    /// # use arrow::datatypes::Schema;
    375
    /// # use arrow::datatypes::Field;
    376
    ///
    377
    /// #[derive(Debug, Clone, PartialEq, Eq, Hash)]
    378
    /// struct GeoMeanUdf {
    379
    ///   signature: Signature,
    380
    /// }
    381
    ///
    382
    /// impl GeoMeanUdf {
    383
    ///   fn new() -> Self {
    384
    ///     Self {
    385
    ///       signature: Signature::uniform(1, vec![DataType::Float64], Volatility::Immutable),
    386
    ///      }
    387
    ///   }
    388
    /// }
    389
    ///
    390
    /// static DOCUMENTATION: LazyLock<Documentation> = LazyLock::new(|| {
    391
    ///         Documentation::builder(DOC_SECTION_AGGREGATE, "calculates a geometric mean", "geo_mean(2.0)")
    392
    ///             .with_argument("arg1", "The Float64 number for the geometric mean")
    393
    ///             .build()
    394
    ///     });
    395
    ///
    396
    /// fn get_doc() -> &'static Documentation {
    397
    ///     &DOCUMENTATION
    398
    /// }
    399
    ///
    400
    /// /// Implement the AggregateUDFImpl trait for GeoMeanUdf
    401
    /// impl AggregateUDFImpl for GeoMeanUdf {
    402
    ///    fn as_any(&self) -> &dyn Any { self }
    403
    ///    fn name(&self) -> &str { "geo_mean" }
    404
    ///    fn signature(&self) -> &Signature { &self.signature }
    405
    ///    fn return_type(&self, args: &[DataType]) -> Result<DataType> {
    406
    ///      if !matches!(args.get(0), Some(&DataType::Float64)) {
    407
    ///        return plan_err!("geo_mean only accepts Float64 arguments");
    408
    ///      }
    409
    ///      Ok(DataType::Float64)
    410
    ///    }
    411
    ///    // This is the accumulator factory; DataFusion uses it to create new accumulators.
    412
    ///    fn accumulator(&self, _acc_args: AccumulatorArgs) -> Result<Box<dyn Accumulator>> { unimplemented!() }
    413
    ///    fn state_fields(&self, args: StateFieldsArgs) -> Result<Vec<FieldRef>> {
    414
    ///        Ok(vec![
    415
    ///             Arc::new(args.return_field.as_ref().clone().with_name("value")),
    416
    ///             Arc::new(Field::new("ordering", DataType::UInt32, true))
    417
    ///        ])
    418
    ///    }
    419
    ///    fn documentation(&self) -> Option<&Documentation> {
    420
    ///        Some(get_doc())
    421
    ///    }
    422
    /// }
    423
    ///
    424
    /// // Create a new AggregateUDF from the implementation
    425
    /// let geometric_mean = AggregateUDF::from(GeoMeanUdf::new());
    426
    ///
    427
    /// // Call the function `geo_mean(col)`
    428
    /// let expr = geometric_mean.call(vec![col("a")]);
    429
    /// ```
    430
    pub trait AggregateUDFImpl: Debug + DynEq + DynHash + Send + Sync {
    431
        /// Returns this object as an [`Any`] trait object
    432
        fn as_any(&self) -> &dyn Any;
    433
    434
        /// Returns this function's name
    435
        fn name(&self) -> &str;
    436
    437
        /// Returns any aliases (alternate names) for this function.
    438
        ///
    439
        /// Note: `aliases` should only include names other than [`Self::name`].
    440
        /// Defaults to `[]` (no aliases)
    441
    0
        fn aliases(&self) -> &[String] {
    442
    0
            &[]
    443
    0
        }
    444
    445
        /// Returns the name of the column this expression would create
    446
        ///
    447
        /// See [`Expr::schema_name`] for details
    448
        ///
    449
        /// Example of schema_name: count(DISTINCT column1) FILTER (WHERE column2 > 10) ORDER BY [..]
    450
    0
        fn schema_name(&self, params: &AggregateFunctionParams) -> Result<String> {
    451
    0
            udaf_default_schema_name(self, params)
    452
    0
        }
    453
    454
        /// Returns a human readable expression.
    455
        ///
    456
        /// See [`Expr::human_display`] for details.
    457
    0
        fn human_display(&self, params: &AggregateFunctionParams) -> Result<String> {
    458
    0
            udaf_default_human_display(self, params)
    459
    0
        }
    460
    461
        /// Returns the name of the column this expression would create
    462
        ///
    463
        /// See [`Expr::schema_name`] for details
    464
        ///
    465
        /// Different from `schema_name` in that it is used for window aggregate function
    466
        ///
    467
        /// Example of schema_name: count(DISTINCT column1) FILTER (WHERE column2 > 10) [PARTITION BY [..]] [ORDER BY [..]]
    468
    0
        fn window_function_schema_name(
    469
    0
            &self,
    470
    0
            params: &WindowFunctionParams,
    471
    0
        ) -> Result<String> {
    472
    0
            udaf_default_window_function_schema_name(self, params)
    473
    0
        }
    474
    475
        /// Returns the user-defined display name of function, given the arguments
    476
        ///
    477
        /// This can be used to customize the output column name generated by this
    478
        /// function.
    479
        ///
    480
        /// Defaults to `function_name([DISTINCT] column1, column2, ..) [null_treatment] [filter] [order_by [..]]`
    481
    0
        fn display_name(&self, params: &AggregateFunctionParams) -> Result<String> {
    482
    0
            udaf_default_display_name(self, params)
    483
    0
        }
    484
    485
        /// Returns the user-defined display name of function, given the arguments
    486
        ///
    487
        /// This can be used to customize the output column name generated by this
    488
        /// function.
    489
        ///
    490
        /// Different from `display_name` in that it is used for window aggregate function
    491
        ///
    492
        /// Defaults to `function_name([DISTINCT] column1, column2, ..) [null_treatment] [partition by [..]] [order_by [..]]`
    493
    0
        fn window_function_display_name(
    494
    0
            &self,
    495
    0
            params: &WindowFunctionParams,
    496
    0
        ) -> Result<String> {
    497
    0
            udaf_default_window_function_display_name(self, params)
    498
    0
        }
    499
    500
        /// Returns the function's [`Signature`] for information about what input
    501
        /// types are accepted and the function's Volatility.
    502
        fn signature(&self) -> &Signature;
    503
    504
        /// What [`DataType`] will be returned by this function, given the types of
    505
        /// the arguments
    506
        fn return_type(&self, arg_types: &[DataType]) -> Result<DataType>;
    507
    508
        /// What type will be returned by this function, given the arguments?
    509
        ///
    510
        /// By default, this function calls [`Self::return_type`] with the
    511
        /// types of each argument.
    512
        ///
    513
        /// # Notes
    514
        ///
    515
        /// Most UDFs should implement [`Self::return_type`] and not this
    516
        /// function as the output type for most functions only depends on the types
    517
        /// of their inputs (e.g. `sum(f64)` is always `f64`).
    518
        ///
    519
        /// This function can be used for more advanced cases such as:
    520
        ///
    521
        /// 1. specifying nullability
    522
        /// 2. return types based on the **values** of the arguments (rather than
    523
        ///    their **types**.
    524
        /// 3. return types based on metadata within the fields of the inputs
    525
    0
        fn return_field(&self, arg_fields: &[FieldRef]) -> Result<FieldRef> {
    526
    0
            udaf_default_return_field(self, arg_fields)
    527
    0
        }
    528
    529
        /// Whether the aggregate function is nullable.
    530
        ///
    531
        /// Nullable means that the function could return `null` for any inputs.
    532
        /// For example, aggregate functions like `COUNT` always return a non null value
    533
        /// but others like `MIN` will return `NULL` if there is nullable input.
    534
        /// Note that if the function is declared as *not* nullable, make sure the [`AggregateUDFImpl::default_value`] is `non-null`
    535
    0
        fn is_nullable(&self) -> bool {
    536
    0
            true
    537
    0
        }
    538
    539
        /// Return a new [`Accumulator`] that aggregates values for a specific
    540
        /// group during query execution.
    541
        ///
    542
        /// acc_args: [`AccumulatorArgs`] contains information about how the
    543
        /// aggregate function was called.
    544
        fn accumulator(&self, acc_args: AccumulatorArgs) -> Result<Box<dyn Accumulator>>;
    545
    546
        /// Return the fields used to store the intermediate state of this accumulator.
    547
        ///
    548
        /// See [`Accumulator::state`] for background information.
    549
        ///
    550
        /// args:  [`StateFieldsArgs`] contains arguments passed to the
    551
        /// aggregate function's accumulator.
    552
        ///
    553
        /// # Notes:
    554
        ///
    555
        /// The default implementation returns a single state field named `name`
    556
        /// with the same type as `value_type`. This is suitable for aggregates such
    557
        /// as `SUM` or `MIN` where partial state can be combined by applying the
    558
        /// same aggregate.
    559
        ///
    560
        /// For aggregates such as `AVG` where the partial state is more complex
    561
        /// (e.g. a COUNT and a SUM), this method is used to define the additional
    562
        /// fields.
    563
        ///
    564
        /// The name of the fields must be unique within the query and thus should
    565
        /// be derived from `name`. See [`format_state_name`] for a utility function
    566
        /// to generate a unique name.
    567
    0
        fn state_fields(&self, args: StateFieldsArgs) -> Result<Vec<FieldRef>> {
    568
    0
            let fields = vec![args
    569
    0
                .return_field
    570
    0
                .as_ref()
    571
    0
                .clone()
    572
    0
                .with_name(format_state_name(args.name, "value"))];
    573
    574
    0
            Ok(fields
    575
    0
                .into_iter()
    576
    0
                .map(Arc::new)
    577
    0
                .chain(args.ordering_fields.to_vec())
    578
    0
                .collect())
    579
    0
        }
    580
    581
        /// If the aggregate expression has a specialized
    582
        /// [`GroupsAccumulator`] implementation. If this returns true,
    583
        /// `[Self::create_groups_accumulator]` will be called.
    584
        ///
    585
        /// # Notes
    586
        ///
    587
        /// Even if this function returns true, DataFusion will still use
    588
        /// [`Self::accumulator`] for certain queries, such as when this aggregate is
    589
        /// used as a window function or when there no GROUP BY columns in the
    590
        /// query.
    591
    0
        fn groups_accumulator_supported(&self, _args: AccumulatorArgs) -> bool {
    592
    0
            false
    593
    0
        }
    594
    595
        /// Return a specialized [`GroupsAccumulator`] that manages state
    596
        /// for all groups.
    597
        ///
    598
        /// For maximum performance, a [`GroupsAccumulator`] should be
    599
        /// implemented in addition to [`Accumulator`].
    600
    0
        fn create_groups_accumulator(
    601
    0
            &self,
    602
    0
            _args: AccumulatorArgs,
    603
    0
        ) -> Result<Box<dyn GroupsAccumulator>> {
    604
    0
            not_impl_err!("GroupsAccumulator hasn't been implemented for {self:?} yet")
    605
    0
        }
    606
    607
        /// Sliding accumulator is an alternative accumulator that can be used for
    608
        /// window functions. It has retract method to revert the previous update.
    609
        ///
    610
        /// See [retract_batch] for more details.
    611
        ///
    612
        /// [retract_batch]: Accumulator::retract_batch
    613
    0
        fn create_sliding_accumulator(
    614
    0
            &self,
    615
    0
            args: AccumulatorArgs,
    616
    0
        ) -> Result<Box<dyn Accumulator>> {
    617
    0
            self.accumulator(args)
    618
    0
        }
    619
    620
        /// Sets the indicator whether ordering requirements of the AggregateUDFImpl is
    621
        /// satisfied by its input. If this is not the case, UDFs with order
    622
        /// sensitivity `AggregateOrderSensitivity::Beneficial` can still produce
    623
        /// the correct result with possibly more work internally.
    624
        ///
    625
        /// # Returns
    626
        ///
    627
        /// Returns `Ok(Some(updated_udf))` if the process completes successfully.
    628
        /// If the expression can benefit from existing input ordering, but does
    629
        /// not implement the method, returns an error. Order insensitive and hard
    630
        /// requirement aggregators return `Ok(None)`.
    631
    0
        fn with_beneficial_ordering(
    632
    0
            self: Arc<Self>,
    633
    0
            _beneficial_ordering: bool,
    634
    0
        ) -> Result<Option<Arc<dyn AggregateUDFImpl>>> {
    635
    0
            if self.order_sensitivity().is_beneficial() {
    636
    0
                return exec_err!(
    637
    0
                    "Should implement with satisfied for aggregator :{:?}",
    638
    0
                    self.name()
    639
                );
    640
    0
            }
    641
    0
            Ok(None)
    642
    0
        }
    643
    644
        /// Gets the order sensitivity of the UDF. See [`AggregateOrderSensitivity`]
    645
        /// for possible options.
    646
    0
        fn order_sensitivity(&self) -> AggregateOrderSensitivity {
    647
            // We have hard ordering requirements by default, meaning that order
    648
            // sensitive UDFs need their input orderings to satisfy their ordering
    649
            // requirements to generate correct results.
    650
    0
            AggregateOrderSensitivity::HardRequirement
    651
    0
        }
    652
    653
        /// Optionally apply per-UDaF simplification / rewrite rules.
    654
        ///
    655
        /// This can be used to apply function specific simplification rules during
    656
        /// optimization (e.g. `arrow_cast` --> `Expr::Cast`). The default
    657
        /// implementation does nothing.
    658
        ///
    659
        /// Note that DataFusion handles simplifying arguments and  "constant
    660
        /// folding" (replacing a function call with constant arguments such as
    661
        /// `my_add(1,2) --> 3` ). Thus, there is no need to implement such
    662
        /// optimizations manually for specific UDFs.
    663
        ///
    664
        /// # Returns
    665
        ///
    666
        /// [None] if simplify is not defined or,
    667
        ///
    668
        /// Or, a closure with two arguments:
    669
        /// * 'aggregate_function': [AggregateFunction] for which simplified has been invoked
    670
        /// * 'info': [crate::simplify::SimplifyInfo]
    671
        ///
    672
        /// closure returns simplified [Expr] or an error.
    673
        ///
    674
        /// # Notes
    675
        ///
    676
        /// The returned expression must have the same schema as the original
    677
        /// expression, including both the data type and nullability. For example,
    678
        /// if the original expression is nullable, the returned expression must
    679
        /// also be nullable, otherwise it may lead to schema verification errors
    680
        /// later in query planning.
    681
    0
        fn simplify(&self) -> Option<AggregateFunctionSimplification> {
    682
    0
            None
    683
    0
        }
    684
    685
        /// Returns the reverse expression of the aggregate function.
    686
    0
        fn reverse_expr(&self) -> ReversedUDAF {
    687
    0
            ReversedUDAF::NotSupported
    688
    0
        }
    689
    690
        /// Coerce arguments of a function call to types that the function can evaluate.
    691
        ///
    692
        /// This function is only called if [`AggregateUDFImpl::signature`] returns [`crate::TypeSignature::UserDefined`]. Most
    693
        /// UDAFs should return one of the other variants of `TypeSignature` which handle common
    694
        /// cases
    695
        ///
    696
        /// See the [type coercion module](crate::type_coercion)
    697
        /// documentation for more details on type coercion
    698
        ///
    699
        /// For example, if your function requires a floating point arguments, but the user calls
    700
        /// it like `my_func(1::int)` (aka with `1` as an integer), coerce_types could return `[DataType::Float64]`
    701
        /// to ensure the argument was cast to `1::double`
    702
        ///
    703
        /// # Parameters
    704
        /// * `arg_types`: The argument types of the arguments  this function with
    705
        ///
    706
        /// # Return value
    707
        /// A Vec the same length as `arg_types`. DataFusion will `CAST` the function call
    708
        /// arguments to these specific types.
    709
    0
        fn coerce_types(&self, _arg_types: &[DataType]) -> Result<Vec<DataType>> {
    710
    0
            not_impl_err!("Function {} does not implement coerce_types", self.name())
    711
    0
        }
    712
    713
        /// If this function is max, return true
    714
        /// If the function is min, return false
    715
        /// Otherwise return None (the default)
    716
        ///
    717
        ///
    718
        /// Note: this is used to use special aggregate implementations in certain conditions
    719
    0
        fn is_descending(&self) -> Option<bool> {
    720
    0
            None
    721
    0
        }
    722
    723
        /// Return the value of this aggregate function if it can be determined
    724
        /// entirely from statistics and arguments.
    725
        ///
    726
        /// Using a [`ScalarValue`] rather than a runtime computation can significantly
    727
        /// improving query performance.
    728
        ///
    729
        /// For example, if the minimum value of column `x` is known to be `42` from
    730
        /// statistics, then the aggregate `MIN(x)` should return `Some(ScalarValue(42))`
    731
    0
        fn value_from_stats(&self, _statistics_args: &StatisticsArgs) -> Option<ScalarValue> {
    732
    0
            None
    733
    0
        }
    734
    735
        /// Returns default value of the function given the input is all `null`.
    736
        ///
    737
        /// Most of the aggregate function return Null if input is Null,
    738
        /// while `count` returns 0 if input is Null
    739
    0
        fn default_value(&self, data_type: &DataType) -> Result<ScalarValue> {
    740
    0
            ScalarValue::try_from(data_type)
    741
    0
        }
    742
    743
        /// If this function supports `[IGNORE NULLS | RESPECT NULLS]` SQL clause,
    744
        /// return `true`. Otherwise, return `false` which will cause an error to be
    745
        /// raised during SQL parsing if these clauses are detected for this function.
    746
        ///
    747
        /// Functions which implement this as `true` are expected to handle the resulting
    748
        /// null handling config present in [`AccumulatorArgs`], `ignore_nulls`.
    749
    0
        fn supports_null_handling_clause(&self) -> bool {
    750
    0
            false
    751
    0
        }
    752
    753
        /// If this function supports the `WITHIN GROUP (ORDER BY column [ASC|DESC])`
    754
        /// SQL syntax, return `true`. Otherwise, return `false` (default) which will
    755
        /// cause an error when parsing SQL where this syntax is detected for this
    756
        /// function.
    757
        ///
    758
        /// This function should return `true` for ordered-set aggregate functions
    759
        /// only.
    760
        ///
    761
        /// # Ordered-set aggregate functions
    762
        ///
    763
        /// Ordered-set aggregate functions allow specifying a sort order that affects
    764
        /// how the function calculates its result, unlike other aggregate functions
    765
        /// like `sum` or `count`. For example, `percentile_cont` is an ordered-set
    766
        /// aggregate function that calculates the exact percentile value from a list
    767
        /// of values; the output of calculating the `0.75` percentile depends on if
    768
        /// you're calculating on an ascending or descending list of values.
    769
        ///
    770
        /// An example of how an ordered-set aggregate function is called with the
    771
        /// `WITHIN GROUP` SQL syntax:
    772
        ///
    773
        /// ```sql
    774
        /// -- Ascending
    775
        /// SELECT percentile_cont(0.75) WITHIN GROUP (ORDER BY c1 ASC) FROM table;
    776
        /// -- Default ordering is ascending if not explicitly specified
    777
        /// SELECT percentile_cont(0.75) WITHIN GROUP (ORDER BY c1) FROM table;
    778
        /// -- Descending
    779
        /// SELECT percentile_cont(0.75) WITHIN GROUP (ORDER BY c1 DESC) FROM table;
    780
        /// ```
    781
        ///
    782
        /// This calculates the `0.75` percentile of the column `c1` from `table`,
    783
        /// according to the specific ordering. The column specified in the `WITHIN GROUP`
    784
        /// ordering clause is taken as the column to calculate values on; specifying
    785
        /// the `WITHIN GROUP` clause is optional so these queries are equivalent:
    786
        ///
    787
        /// ```sql
    788
        /// -- If no WITHIN GROUP is specified then default ordering is implementation
    789
        /// -- dependent; in this case ascending for percentile_cont
    790
        /// SELECT percentile_cont(c1, 0.75) FROM table;
    791
        /// SELECT percentile_cont(0.75) WITHIN GROUP (ORDER BY c1 ASC) FROM table;
    792
        /// ```
    793
        ///
    794
        /// Aggregate UDFs can define their default ordering if the function is called
    795
        /// without the `WITHIN GROUP` clause, though a default of ascending is the
    796
        /// standard practice.
    797
        ///
    798
        /// Ordered-set aggregate function implementations are responsible for handling
    799
        /// the input sort order themselves (e.g. `percentile_cont` must buffer and
    800
        /// sort the values internally). That is, DataFusion does not introduce any
    801
        /// kind of sort into the plan for these functions with this syntax.
    802
    0
        fn supports_within_group_clause(&self) -> bool {
    803
    0
            false
    804
    0
        }
    805
    806
        /// Returns the documentation for this Aggregate UDF.
    807
        ///
    808
        /// Documentation can be accessed programmatically as well as
    809
        /// generating publicly facing documentation.
    810
    0
        fn documentation(&self) -> Option<&Documentation> {
    811
    0
            None
    812
    0
        }
    813
    814
        /// Indicates whether the aggregation function is monotonic as a set
    815
        /// function. See [`SetMonotonicity`] for details.
    816
    0
        fn set_monotonicity(&self, _data_type: &DataType) -> SetMonotonicity {
    817
    0
            SetMonotonicity::NotMonotonic
    818
    0
        }
    819
    }
    820
    821
    impl PartialEq for dyn AggregateUDFImpl {
    822
    0
        fn eq(&self, other: &Self) -> bool {
    823
    0
            self.dyn_eq(other.as_any())
    824
    0
        }
    825
    }
    826
    827
    impl PartialOrd for dyn AggregateUDFImpl {
    828
    0
        fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
    829
    0
            match self.name().partial_cmp(other.name()) {
    830
    0
                Some(Ordering::Equal) => self.signature().partial_cmp(other.signature()),
    831
    0
                cmp => cmp,
    832
            }
    833
            // TODO (https://github.com/apache/datafusion/issues/17477) avoid recomparing all fields
    834
    0
            .filter(|cmp| *cmp != Ordering::Equal || self == other)
    835
    0
        }
    836
    }
    837
    838
    /// Encapsulates default implementation of [`AggregateUDFImpl::schema_name`].
    839
    0
    pub fn udaf_default_schema_name<F: AggregateUDFImpl + ?Sized>(
    840
    0
        func: &F,
    841
    0
        params: &AggregateFunctionParams,
    842
    0
    ) -> Result<String> {
    843
        let AggregateFunctionParams {
    844
    0
            args,
    845
    0
            distinct,
    846
    0
            filter,
    847
    0
            order_by,
    848
    0
            null_treatment,
    849
    0
        } = params;
    850
    851
        // exclude the first function argument(= column) in ordered set aggregate function,
    852
        // because it is duplicated with the WITHIN GROUP clause in schema name.
    853
    0
        let args = if func.supports_within_group_clause() && !order_by.is_empty() {
    854
    0
            &args[1..]
    855
        } else {
    856
    0
            &args[..]
    857
        };
    858
    859
    0
        let mut schema_name = String::new();
    860
    861
    0
        schema_name.write_fmt(format_args!(
    862
    0
            "{}({}{})",
    863
    0
            func.name(),
    864
    0
            if *distinct { "DISTINCT " } else { "" },
    865
    0
            schema_name_from_exprs_comma_separated_without_space(args)?
    866
    0
        ))?;
    867
    868
    0
        if let Some(null_treatment) = null_treatment {
    869
    0
            schema_name.write_fmt(format_args!(" {null_treatment}"))?;
    870
    0
        }
    871
    872
    0
        if let Some(filter) = filter {
    873
    0
            schema_name.write_fmt(format_args!(" FILTER (WHERE {filter})"))?;
    874
    0
        };
    875
    876
    0
        if !order_by.is_empty() {
    877
    0
            let clause = match func.supports_within_group_clause() {
    878
    0
                true => "WITHIN GROUP",
    879
    0
                false => "ORDER BY",
    880
            };
    881
    882
    0
            schema_name.write_fmt(format_args!(
    883
    0
                " {} [{}]",
    884
                clause,
    885
    0
                schema_name_from_sorts(order_by)?
    886
    0
            ))?;
    887
    0
        };
    888
    889
    0
        Ok(schema_name)
    890
    0
    }
    891
    892
    /// Encapsulates default implementation of [`AggregateUDFImpl::human_display`].
    893
    0
    pub fn udaf_default_human_display<F: AggregateUDFImpl + ?Sized>(
    894
    0
        func: &F,
    895
    0
        params: &AggregateFunctionParams,
    896
    0
    ) -> Result<String> {
    897
        let AggregateFunctionParams {
    898
    0
            args,
    899
    0
            distinct,
    900
    0
            filter,
    901
    0
            order_by,
    902
    0
            null_treatment,
    903
    0
        } = params;
    904
    905
    0
        let mut schema_name = String::new();
    906
    907
    0
        schema_name.write_fmt(format_args!(
    908
    0
            "{}({}{})",
    909
    0
            func.name(),
    910
    0
            if *distinct { "DISTINCT " } else { "" },
    911
    0
            ExprListDisplay::comma_separated(args.as_slice())
    912
    0
        ))?;
    913
    914
    0
        if let Some(null_treatment) = null_treatment {
    915
    0
            schema_name.write_fmt(format_args!(" {null_treatment}"))?;
    916
    0
        }
    917
    918
    0
        if let Some(filter) = filter {
    919
    0
            schema_name.write_fmt(format_args!(" FILTER (WHERE {filter})"))?;
    920
    0
        };
    921
    922
    0
        if !order_by.is_empty() {
    923
    0
            schema_name.write_fmt(format_args!(
    924
    0
                " ORDER BY [{}]",
    925
    0
                schema_name_from_sorts(order_by)?
    926
    0
            ))?;
    927
    0
        };
    928
    929
    0
        Ok(schema_name)
    930
    0
    }
    931
    932
    /// Encapsulates default implementation of [`AggregateUDFImpl::window_function_schema_name`].
    933
    0
    pub fn udaf_default_window_function_schema_name<F: AggregateUDFImpl + ?Sized>(
    934
    0
        func: &F,
    935
    0
        params: &WindowFunctionParams,
    936
    0
    ) -> Result<String> {
    937
        let WindowFunctionParams {
    938
    0
            args,
    939
    0
            partition_by,
    940
    0
            order_by,
    941
    0
            window_frame,
    942
    0
            filter,
    943
    0
            null_treatment,
    944
    0
            distinct,
    945
    0
        } = params;
    946
    947
    0
        let mut schema_name = String::new();
    948
    949
        // Inject DISTINCT into the schema name when requested
    950
    0
        if *distinct {
    951
    0
            schema_name.write_fmt(format_args!(
    952
    0
                "{}(DISTINCT {})",
    953
    0
                func.name(),
    954
    0
                schema_name_from_exprs(args)?
    955
    0
            ))?;
    956
        } else {
    957
    0
            schema_name.write_fmt(format_args!(
    958
    0
                "{}({})",
    959
    0
                func.name(),
    960
    0
                schema_name_from_exprs(args)?
    961
    0
            ))?;
    962
        }
    963
    964
    0
        if let Some(null_treatment) = null_treatment {
    965
    0
            schema_name.write_fmt(format_args!(" {null_treatment}"))?;
    966
    0
        }
    967
    968
    0
        if let Some(filter) = filter {
    969
    0
            schema_name.write_fmt(format_args!(" FILTER (WHERE {filter})"))?;
    970
    0
        }
    971
    972
    0
        if !partition_by.is_empty() {
    973
    0
            schema_name.write_fmt(format_args!(
    974
    0
                " PARTITION BY [{}]",
    975
    0
                schema_name_from_exprs(partition_by)?
    976
    0
            ))?;
    977
    0
        }
    978
    979
    0
        if !order_by.is_empty() {
    980
    0
            schema_name.write_fmt(format_args!(
    981
    0
                " ORDER BY [{}]",
    982
    0
                schema_name_from_sorts(order_by)?
    983
    0
            ))?;
    984
    0
        }
    985
    986
    0
        schema_name.write_fmt(format_args!(" {window_frame}"))?;
    987
    988
    0
        Ok(schema_name)
    989
    0
    }
    990
    991
    /// Encapsulates default implementation of [`AggregateUDFImpl::display_name`].
    992
    0
    pub fn udaf_default_display_name<F: AggregateUDFImpl + ?Sized>(
    993
    0
        func: &F,
    994
    0
        params: &AggregateFunctionParams,
    995
    0
    ) -> Result<String> {
    996
        let AggregateFunctionParams {
    997
    0
            args,
    998
    0
            distinct,
    999
    0
            filter,
    1000
    0
            order_by,
    1001
    0
            null_treatment,
    1002
    0
        } = params;
    1003
    1004
    0
        let mut display_name = String::new();
    1005
    1006
    0
        display_name.write_fmt(format_args!(
    1007
    0
            "{}({}{})",
    1008
    0
            func.name(),
    1009
    0
            if *distinct { "DISTINCT " } else { "" },
    1010
    0
            expr_vec_fmt!(args)
    1011
    0
        ))?;
    1012
    1013
    0
        if let Some(nt) = null_treatment {
    1014
    0
            display_name.write_fmt(format_args!(" {nt}"))?;
    1015
    0
        }
    1016
    0
        if let Some(fe) = filter {
    1017
    0
            display_name.write_fmt(format_args!(" FILTER (WHERE {fe})"))?;
    1018
    0
        }
    1019
    0
        if !order_by.is_empty() {
    1020
    0
            display_name.write_fmt(format_args!(
    1021
    0
                " ORDER BY [{}]",
    1022
    0
                order_by
    1023
    0
                    .iter()
    1024
    0
                    .map(|o| format!("{o}"))
    1025
    0
                    .collect::<Vec<String>>()
    1026
    0
                    .join(", ")
    1027
    0
            ))?;
    1028
    0
        }
    1029
    1030
    0
        Ok(display_name)
    1031
    0
    }
    1032
    1033
    /// Encapsulates default implementation of [`AggregateUDFImpl::window_function_display_name`].
    1034
    0
    pub fn udaf_default_window_function_display_name<F: AggregateUDFImpl + ?Sized>(
    1035
    0
        func: &F,
    1036
    0
        params: &WindowFunctionParams,
    1037
    0
    ) -> Result<String> {
    1038
        let WindowFunctionParams {
    1039
    0
            args,
    1040
    0
            partition_by,
    1041
    0
            order_by,
    1042
    0
            window_frame,
    1043
    0
            filter,
    1044
    0
            null_treatment,
    1045
    0
            distinct,
    1046
    0
        } = params;
    1047
    1048
    0
        let mut display_name = String::new();
    1049
    1050
    0
        if *distinct {
    1051
    0
            display_name.write_fmt(format_args!(
    1052
    0
                "{}(DISTINCT {})",
    1053
    0
                func.name(),
    1054
    0
                expr_vec_fmt!(args)
    1055
    0
            ))?;
    1056
        } else {
    1057
    0
            display_name.write_fmt(format_args!(
    1058
    0
                "{}({})",
    1059
    0
                func.name(),
    1060
    0
                expr_vec_fmt!(args)
    1061
    0
            ))?;
    1062
        }
    1063
    1064
    0
        if let Some(null_treatment) = null_treatment {
    1065
    0
            display_name.write_fmt(format_args!(" {null_treatment}"))?;
    1066
    0
        }
    1067
    1068
    0
        if let Some(fe) = filter {
    1069
    0
            display_name.write_fmt(format_args!(" FILTER (WHERE {fe})"))?;
    1070
    0
        }
    1071
    1072
    0
        if !partition_by.is_empty() {
    1073
    0
            display_name.write_fmt(format_args!(
    1074
    0
                " PARTITION BY [{}]",
    1075
    0
                expr_vec_fmt!(partition_by)
    1076
    0
            ))?;
    1077
    0
        }
    1078
    1079
    0
        if !order_by.is_empty() {
    1080
    0
            display_name
    1081
    0
                .write_fmt(format_args!(" ORDER BY [{}]", expr_vec_fmt!(order_by)))?;
    1082
    0
        };
    1083
    1084
    0
        display_name.write_fmt(format_args!(
    1085
    0
            " {} BETWEEN {} AND {}",
    1086
            window_frame.units, window_frame.start_bound, window_frame.end_bound
    1087
    0
        ))?;
    1088
    1089
    0
        Ok(display_name)
    1090
    0
    }
    1091
    1092
    /// Encapsulates default implementation of [`AggregateUDFImpl::return_field`].
    1093
    0
    pub fn udaf_default_return_field<F: AggregateUDFImpl + ?Sized>(
    1094
    0
        func: &F,
    1095
    0
        arg_fields: &[FieldRef],
    1096
    0
    ) -> Result<FieldRef> {
    1097
    0
        let arg_types: Vec<_> = arg_fields.iter().map(|f| f.data_type()).cloned().collect();
    1098
    0
        let data_type = func.return_type(&arg_types)?;
    1099
    1100
    0
        Ok(Arc::new(Field::new(
    1101
    0
            func.name(),
    1102
    0
            data_type,
    1103
    0
            func.is_nullable(),
    1104
    0
        )))
    1105
    0
    }
    1106
    1107
    pub enum ReversedUDAF {
    1108
        /// The expression is the same as the original expression, like SUM, COUNT
    1109
        Identical,
    1110
        /// The expression does not support reverse calculation
    1111
        NotSupported,
    1112
        /// The expression is different from the original expression
    1113
        Reversed(Arc<AggregateUDF>),
    1114
    }
    1115
    1116
    /// AggregateUDF that adds an alias to the underlying function. It is better to
    1117
    /// implement [`AggregateUDFImpl`], which supports aliases, directly if possible.
    1118
    #[derive(Debug, PartialEq, Eq, Hash)]
    1119
    struct AliasedAggregateUDFImpl {
    1120
        inner: UdfEq<Arc<dyn AggregateUDFImpl>>,
    1121
        aliases: Vec<String>,
    1122
    }
    1123
    1124
    impl AliasedAggregateUDFImpl {
    1125
    0
        pub fn new(
    1126
    0
            inner: Arc<dyn AggregateUDFImpl>,
    1127
    0
            new_aliases: impl IntoIterator<Item = &'static str>,
    1128
    0
        ) -> Self {
    1129
    0
            let mut aliases = inner.aliases().to_vec();
    1130
    0
            aliases.extend(new_aliases.into_iter().map(|s| s.to_string()));
    1131
    1132
    0
            Self {
    1133
    0
                inner: inner.into(),
    1134
    0
                aliases,
    1135
    0
            }
    1136
    0
        }
    1137
    }
    1138
    1139
    #[warn(clippy::missing_trait_methods)] // Delegates, so it should implement every single trait method
    1140
    impl AggregateUDFImpl for AliasedAggregateUDFImpl {
    1141
    0
        fn as_any(&self) -> &dyn Any {
    1142
    0
            self
    1143
    0
        }
    1144
    1145
    0
        fn name(&self) -> &str {
    1146
    0
            self.inner.name()
    1147
    0
        }
    1148
    1149
    0
        fn signature(&self) -> &Signature {
    1150
    0
            self.inner.signature()
    1151
    0
        }
    1152
    1153
    0
        fn return_type(&self, arg_types: &[DataType]) -> Result<DataType> {
    1154
    0
            self.inner.return_type(arg_types)
    1155
    0
        }
    1156
    1157
    0
        fn accumulator(&self, acc_args: AccumulatorArgs) -> Result<Box<dyn Accumulator>> {
    1158
    0
            self.inner.accumulator(acc_args)
    1159
    0
        }
    1160
    1161
    0
        fn aliases(&self) -> &[String] {
    1162
    0
            &self.aliases
    1163
    0
        }
    1164
    1165
    0
        fn schema_name(&self, params: &AggregateFunctionParams) -> Result<String> {
    1166
    0
            self.inner.schema_name(params)
    1167
    0
        }
    1168
    1169
    0
        fn human_display(&self, params: &AggregateFunctionParams) -> Result<String> {
    1170
    0
            self.inner.human_display(params)
    1171
    0
        }
    1172
    1173
    0
        fn window_function_schema_name(
    1174
    0
            &self,
    1175
    0
            params: &WindowFunctionParams,
    1176
    0
        ) -> Result<String> {
    1177
    0
            self.inner.window_function_schema_name(params)
    1178
    0
        }
    1179
    1180
    0
        fn display_name(&self, params: &AggregateFunctionParams) -> Result<String> {
    1181
    0
            self.inner.display_name(params)
    1182
    0
        }
    1183
    1184
    0
        fn window_function_display_name(
    1185
    0
            &self,
    1186
    0
            params: &WindowFunctionParams,
    1187
    0
        ) -> Result<String> {
    1188
    0
            self.inner.window_function_display_name(params)
    1189
    0
        }
    1190
    1191
    0
        fn state_fields(&self, args: StateFieldsArgs) -> Result<Vec<FieldRef>> {
    1192
    0
            self.inner.state_fields(args)
    1193
    0
        }
    1194
    1195
    0
        fn groups_accumulator_supported(&self, args: AccumulatorArgs) -> bool {
    1196
    0
            self.inner.groups_accumulator_supported(args)
    1197
    0
        }
    1198
    1199
    0
        fn create_groups_accumulator(
    1200
    0
            &self,
    1201
    0
            args: AccumulatorArgs,
    1202
    0
        ) -> Result<Box<dyn GroupsAccumulator>> {
    1203
    0
            self.inner.create_groups_accumulator(args)
    1204
    0
        }
    1205
    1206
    0
        fn create_sliding_accumulator(
    1207
    0
            &self,
    1208
    0
            args: AccumulatorArgs,
    1209
    0
        ) -> Result<Box<dyn Accumulator>> {
    1210
    0
            self.inner.accumulator(args)
    1211
    0
        }
    1212
    1213
    0
        fn with_beneficial_ordering(
    1214
    0
            self: Arc<Self>,
    1215
    0
            beneficial_ordering: bool,
    1216
    0
        ) -> Result<Option<Arc<dyn AggregateUDFImpl>>> {
    1217
    0
            Arc::clone(&self.inner)
    1218
    0
                .with_beneficial_ordering(beneficial_ordering)
    1219
    0
                .map(|udf| {
    1220
    0
                    udf.map(|udf| {
    1221
    0
                        Arc::new(AliasedAggregateUDFImpl {
    1222
    0
                            inner: udf.into(),
    1223
    0
                            aliases: self.aliases.clone(),
    1224
    0
                        }) as Arc<dyn AggregateUDFImpl>
    1225
    0
                    })
    1226
    0
                })
    1227
    0
        }
    1228
    1229
    0
        fn order_sensitivity(&self) -> AggregateOrderSensitivity {
    1230
    0
            self.inner.order_sensitivity()
    1231
    0
        }
    1232
    1233
    0
        fn simplify(&self) -> Option<AggregateFunctionSimplification> {
    1234
    0
            self.inner.simplify()
    1235
    0
        }
    1236
    1237
    0
        fn reverse_expr(&self) -> ReversedUDAF {
    1238
    0
            self.inner.reverse_expr()
    1239
    0
        }
    1240
    1241
    0
        fn coerce_types(&self, arg_types: &[DataType]) -> Result<Vec<DataType>> {
    1242
    0
            self.inner.coerce_types(arg_types)
    1243
    0
        }
    1244
    1245
    0
        fn return_field(&self, arg_fields: &[FieldRef]) -> Result<FieldRef> {
    1246
    0
            self.inner.return_field(arg_fields)
    1247
    0
        }
    1248
    1249
    0
        fn is_nullable(&self) -> bool {
    1250
    0
            self.inner.is_nullable()
    1251
    0
        }
    1252
    1253
    0
        fn is_descending(&self) -> Option<bool> {
    1254
    0
            self.inner.is_descending()
    1255
    0
        }
    1256
    1257
    0
        fn value_from_stats(&self, statistics_args: &StatisticsArgs) -> Option<ScalarValue> {
    1258
    0
            self.inner.value_from_stats(statistics_args)
    1259
    0
        }
    1260
    1261
    0
        fn default_value(&self, data_type: &DataType) -> Result<ScalarValue> {
    1262
    0
            self.inner.default_value(data_type)
    1263
    0
        }
    1264
    1265
    0
        fn supports_null_handling_clause(&self) -> bool {
    1266
    0
            self.inner.supports_null_handling_clause()
    1267
    0
        }
    1268
    1269
    0
        fn supports_within_group_clause(&self) -> bool {
    1270
    0
            self.inner.supports_within_group_clause()
    1271
    0
        }
    1272
    1273
    0
        fn set_monotonicity(&self, data_type: &DataType) -> SetMonotonicity {
    1274
    0
            self.inner.set_monotonicity(data_type)
    1275
    0
        }
    1276
    1277
    0
        fn documentation(&self) -> Option<&Documentation> {
    1278
    0
            self.inner.documentation()
    1279
    0
        }
    1280
    }
    1281
    1282
    /// Indicates whether an aggregation function is monotonic as a set
    1283
    /// function. A set function is monotonically increasing if its value
    1284
    /// increases as its argument grows (as a set). Formally, `f` is a
    1285
    /// monotonically increasing set function if `f(S) >= f(T)` whenever `S`
    1286
    /// is a superset of `T`.
    1287
    ///
    1288
    /// For example `COUNT` and `MAX` are monotonically increasing as their
    1289
    /// values always increase (or stay the same) as new values are seen. On
    1290
    /// the other hand, `MIN` is monotonically decreasing as its value always
    1291
    /// decreases or stays the same as new values are seen.
    1292
    #[derive(Debug, Clone, PartialEq)]
    1293
    pub enum SetMonotonicity {
    1294
        /// Aggregate value increases or stays the same as the input set grows.
    1295
        Increasing,
    1296
        /// Aggregate value decreases or stays the same as the input set grows.
    1297
        Decreasing,
    1298
        /// Aggregate value may increase, decrease, or stay the same as the input
    1299
        /// set grows.
    1300
        NotMonotonic,
    1301
    }
    1302
    1303
    #[cfg(test)]
    1304
    mod test {
    1305
        use crate::{AggregateUDF, AggregateUDFImpl};
    1306
        use arrow::datatypes::{DataType, FieldRef};
    1307
        use datafusion_common::Result;
    1308
        use datafusion_expr_common::accumulator::Accumulator;
    1309
        use datafusion_expr_common::signature::{Signature, Volatility};
    1310
        use datafusion_functions_aggregate_common::accumulator::{
    1311
            AccumulatorArgs, StateFieldsArgs,
    1312
        };
    1313
        use std::any::Any;
    1314
        use std::cmp::Ordering;
    1315
        use std::hash::{DefaultHasher, Hash, Hasher};
    1316
    1317
        #[derive(Debug, Clone, PartialEq, Eq, Hash)]
    1318
        struct AMeanUdf {
    1319
            signature: Signature,
    1320
        }
    1321
    1322
        impl AMeanUdf {
    1323
            fn new() -> Self {
    1324
                Self {
    1325
                    signature: Signature::uniform(
    1326
                        1,
    1327
                        vec![DataType::Float64],
    1328
                        Volatility::Immutable,
    1329
                    ),
    1330
                }
    1331
            }
    1332
        }
    1333
    1334
        impl AggregateUDFImpl for AMeanUdf {
    1335
            fn as_any(&self) -> &dyn Any {
    1336
                self
    1337
            }
    1338
            fn name(&self) -> &str {
    1339
                "a"
    1340
            }
    1341
            fn signature(&self) -> &Signature {
    1342
                &self.signature
    1343
            }
    1344
            fn return_type(&self, _args: &[DataType]) -> Result<DataType> {
    1345
                unimplemented!()
    1346
            }
    1347
            fn accumulator(
    1348
                &self,
    1349
                _acc_args: AccumulatorArgs,
    1350
            ) -> Result<Box<dyn Accumulator>> {
    1351
                unimplemented!()
    1352
            }
    1353
            fn state_fields(&self, _args: StateFieldsArgs) -> Result<Vec<FieldRef>> {
    1354
                unimplemented!()
    1355
            }
    1356
        }
    1357
    1358
        #[derive(Debug, Clone, PartialEq, Eq, Hash)]
    1359
        struct BMeanUdf {
    1360
            signature: Signature,
    1361
        }
    1362
        impl BMeanUdf {
    1363
            fn new() -> Self {
    1364
                Self {
    1365
                    signature: Signature::uniform(
    1366
                        1,
    1367
                        vec![DataType::Float64],
    1368
                        Volatility::Immutable,
    1369
                    ),
    1370
                }
    1371
            }
    1372
        }
    1373
    1374
        impl AggregateUDFImpl for BMeanUdf {
    1375
            fn as_any(&self) -> &dyn Any {
    1376
                self
    1377
            }
    1378
            fn name(&self) -> &str {
    1379
                "b"
    1380
            }
    1381
            fn signature(&self) -> &Signature {
    1382
                &self.signature
    1383
            }
    1384
            fn return_type(&self, _args: &[DataType]) -> Result<DataType> {
    1385
                unimplemented!()
    1386
            }
    1387
            fn accumulator(
    1388
                &self,
    1389
                _acc_args: AccumulatorArgs,
    1390
            ) -> Result<Box<dyn Accumulator>> {
    1391
                unimplemented!()
    1392
            }
    1393
            fn state_fields(&self, _args: StateFieldsArgs) -> Result<Vec<FieldRef>> {
    1394
                unimplemented!()
    1395
            }
    1396
        }
    1397
    1398
        #[test]
    1399
        fn test_partial_eq() {
    1400
            let a1 = AggregateUDF::from(AMeanUdf::new());
    1401
            let a2 = AggregateUDF::from(AMeanUdf::new());
    1402
            let eq = a1 == a2;
    1403
            assert!(eq);
    1404
            assert_eq!(a1, a2);
    1405
            assert_eq!(hash(a1), hash(a2));
    1406
        }
    1407
    1408
        #[test]
    1409
        fn test_partial_ord() {
    1410
            // Test validates that partial ord is defined for AggregateUDF using the name and signature,
    1411
            // not intended to exhaustively test all possibilities
    1412
            let a1 = AggregateUDF::from(AMeanUdf::new());
    1413
            let a2 = AggregateUDF::from(AMeanUdf::new());
    1414
            assert_eq!(a1.partial_cmp(&a2), Some(Ordering::Equal));
    1415
    1416
            let b1 = AggregateUDF::from(BMeanUdf::new());
    1417
            assert!(a1 < b1);
    1418
            assert!(!(a1 == b1));
    1419
        }
    1420
    1421
        fn hash<T: Hash>(value: T) -> u64 {
    1422
            let hasher = &mut DefaultHasher::new();
    1423
            value.hash(hasher);
    1424
            hasher.finish()
    1425
        }
    1426
    }