chore: remove as_any from aggregate and window functions (#21209)

## Which issue does this PR close?

This is a follow on to #20812
but treats aggregate and window functions.

## Rationale for this change

This PR reduces the amount of boilerplate code that users need to write
for aggregate and window functions.

## What changes are included in this PR?

Now that we have [trait
upcasting](https://blog.rust-lang.org/2025/04/03/Rust-1.86.0/) since
rust 1.86, we no longer need every implementation of these functions to
have the as_any function that returns &self. This PR makes Any an
supertrait and makes the appropriate casts when necessary.

## Are these changes tested?

Existing unit tests.

## Are there any user-facing changes?

Yes, the users simply need to remove the `as_any` function. The upgrade
guide is updated.

---------

Co-authored-by: Andrew Lamb <andrew@nerdnetworks.org>

Author: timsaucer

datafusion-examples/examples/udf/advanced_udaf.rs

@@ -20,7 +20,7 @@
 use arrow::datatypes::{Field, Schema};
 use datafusion::physical_expr::NullState;
 use datafusion::{arrow::datatypes::DataType, logical_expr::Volatility};
-use std::{any::Any, sync::Arc};
+use std::sync::Arc;
 
 use arrow::array::{
     ArrayRef, AsArray, Float32Array, PrimitiveArray, PrimitiveBuilder, UInt32Array,
@@ -64,11 +64,6 @@ impl GeoMeanUdaf {
 }
 
 impl AggregateUDFImpl for GeoMeanUdaf {
-    /// We implement as_any so that we can downcast the AggregateUDFImpl trait object
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     /// Return the name of this function
     fn name(&self) -> &str {
         "geo_mean"
@@ -387,10 +382,6 @@ impl SimplifiedGeoMeanUdaf {
 }
 
 impl AggregateUDFImpl for SimplifiedGeoMeanUdaf {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "simplified_geo_mean"
     }

datafusion-examples/examples/udf/advanced_udwf.rs

@@ -17,7 +17,7 @@
 
 //! See `main.rs` for how to run it.
 
-use std::{any::Any, sync::Arc};
+use std::sync::Arc;
 
 use arrow::datatypes::Field;
 use arrow::{
@@ -68,11 +68,6 @@ impl SmoothItUdf {
 }
 
 impl WindowUDFImpl for SmoothItUdf {
-    /// We implement as_any so that we can downcast the WindowUDFImpl trait object
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     /// Return the name of this function
     fn name(&self) -> &str {
         "smooth_it"
@@ -176,10 +171,6 @@ impl SimplifySmoothItUdf {
     }
 }
 impl WindowUDFImpl for SimplifySmoothItUdf {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "simplify_smooth_it"
     }

datafusion/core/tests/user_defined/user_defined_aggregates.rs

@@ -18,7 +18,6 @@
 //! This module contains end to end demonstrations of creating
 //! user defined aggregate functions
 
-use std::any::Any;
 use std::collections::HashMap;
 use std::hash::{Hash, Hasher};
 use std::mem::{size_of, size_of_val};
@@ -793,10 +792,6 @@ struct TestGroupsAccumulator {
 }
 
 impl AggregateUDFImpl for TestGroupsAccumulator {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "geo_mean"
     }
@@ -936,10 +931,6 @@ impl MetadataBasedAggregateUdf {
 }
 
 impl AggregateUDFImpl for MetadataBasedAggregateUdf {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         &self.name
     }

datafusion/core/tests/user_defined/user_defined_window_functions.rs

@@ -44,7 +44,6 @@ use datafusion_physical_expr::{
 use std::collections::HashMap;
 use std::hash::{Hash, Hasher};
 use std::{
-    any::Any,
     ops::Range,
     sync::{
         Arc,
@@ -546,10 +545,6 @@ impl OddCounter {
         }
 
         impl WindowUDFImpl for SimpleWindowUDF {
-            fn as_any(&self) -> &dyn Any {
-                self
-            }
-
             fn name(&self) -> &str {
                 "odd_counter"
             }
@@ -675,10 +670,6 @@ impl VariadicWindowUDF {
 }
 
 impl WindowUDFImpl for VariadicWindowUDF {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "variadic_window_udf"
     }
@@ -819,10 +810,6 @@ impl MetadataBasedWindowUdf {
 }
 
 impl WindowUDFImpl for MetadataBasedWindowUdf {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         &self.name
     }

datafusion/expr/src/expr_fn.rs

@@ -43,7 +43,6 @@ use datafusion_common::{Column, Result, ScalarValue, Spans, TableReference, plan
 use datafusion_functions_window_common::field::WindowUDFFieldArgs;
 use datafusion_functions_window_common::partition::PartitionEvaluatorArgs;
 use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
-use std::any::Any;
 use std::collections::HashMap;
 use std::fmt::Debug;
 use std::hash::Hash;
@@ -588,10 +587,6 @@ impl SimpleAggregateUDF {
 }
 
 impl AggregateUDFImpl for SimpleAggregateUDF {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         &self.name
     }
@@ -681,10 +676,6 @@ impl SimpleWindowUDF {
 }
 
 impl WindowUDFImpl for SimpleWindowUDF {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         &self.name
     }

datafusion/expr/src/test/function_stub.rs

@@ -19,8 +19,6 @@
 //!
 //! These are used to avoid a dependence on `datafusion-functions-aggregate` which live in a different crate
 
-use std::any::Any;
-
 use arrow::datatypes::{
     DECIMAL32_MAX_PRECISION, DECIMAL32_MAX_SCALE, DECIMAL64_MAX_PRECISION,
     DECIMAL64_MAX_SCALE, DECIMAL128_MAX_PRECISION, DECIMAL128_MAX_SCALE,
@@ -113,10 +111,6 @@ impl Default for Sum {
 }
 
 impl AggregateUDFImpl for Sum {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "sum"
     }
@@ -245,10 +239,6 @@ impl Count {
 }
 
 impl AggregateUDFImpl for Count {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "COUNT"
     }
@@ -332,10 +322,6 @@ impl Min {
 }
 
 impl AggregateUDFImpl for Min {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "min"
     }
@@ -414,10 +400,6 @@ impl Max {
 }
 
 impl AggregateUDFImpl for Max {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "max"
     }
@@ -476,10 +458,6 @@ impl Default for Avg {
 }
 
 impl AggregateUDFImpl for Avg {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "avg"
     }

datafusion/expr/src/udaf.rs

@@ -84,7 +84,7 @@ pub struct AggregateUDF {
 
 impl PartialEq for AggregateUDF {
     fn eq(&self, other: &Self) -> bool {
-        self.inner.dyn_eq(other.inner.as_any())
+        self.inner.dyn_eq(other.inner.as_ref() as &dyn Any)
     }
 }
 
@@ -415,7 +415,6 @@ where
 ///
 /// /// Implement the AggregateUDFImpl trait for GeoMeanUdf
 /// impl AggregateUDFImpl for GeoMeanUdf {
-///    fn as_any(&self) -> &dyn Any { self }
 ///    fn name(&self) -> &str { "geo_mean" }
 ///    fn signature(&self) -> &Signature { &self.signature }
 ///    fn return_type(&self, args: &[DataType]) -> Result<DataType> {
@@ -443,10 +442,7 @@ where
 /// // Call the function `geo_mean(col)`
 /// let expr = geometric_mean.call(vec![col("a")]);
 /// ```
-pub trait AggregateUDFImpl: Debug + DynEq + DynHash + Send + Sync {
-    /// Returns this object as an [`Any`] trait object
-    fn as_any(&self) -> &dyn Any;
-
+pub trait AggregateUDFImpl: Debug + DynEq + DynHash + Send + Sync + Any {
     /// Returns this function's name
     fn name(&self) -> &str;
 
@@ -913,7 +909,7 @@ pub trait AggregateUDFImpl: Debug + DynEq + DynHash + Send + Sync {
 
 impl PartialEq for dyn AggregateUDFImpl {
     fn eq(&self, other: &Self) -> bool {
-        self.dyn_eq(other.as_any())
+        self.dyn_eq(other)
     }
 }
 
@@ -1231,10 +1227,6 @@ impl AliasedAggregateUDFImpl {
 
 #[warn(clippy::missing_trait_methods)] // Delegates, so it should implement every single trait method
 impl AggregateUDFImpl for AliasedAggregateUDFImpl {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         self.inner.name()
     }
@@ -1415,7 +1407,6 @@ mod test {
     use datafusion_functions_aggregate_common::accumulator::{
         AccumulatorArgs, StateFieldsArgs,
     };
-    use std::any::Any;
     use std::cmp::Ordering;
     use std::hash::{DefaultHasher, Hash, Hasher};
 
@@ -1437,9 +1428,6 @@ mod test {
     }
 
     impl AggregateUDFImpl for AMeanUdf {
-        fn as_any(&self) -> &dyn Any {
-            self
-        }
         fn name(&self) -> &str {
             "a"
         }
@@ -1477,9 +1465,6 @@ mod test {
     }
 
     impl AggregateUDFImpl for BMeanUdf {
-        fn as_any(&self) -> &dyn Any {
-            self
-        }
         fn name(&self) -> &str {
             "b"
         }

datafusion/expr/src/udf_eq.rs

@@ -96,7 +96,7 @@ impl UdfPointer for Arc<dyn ScalarUDFImpl + '_> {
 
 impl UdfPointer for Arc<dyn AggregateUDFImpl + '_> {
     fn equals(&self, other: &(dyn AggregateUDFImpl + '_)) -> bool {
-        self.as_ref().dyn_eq(other.as_any())
+        self.as_ref().dyn_eq(other)
     }
 
     fn hash_value(&self) -> u64 {
@@ -108,7 +108,7 @@ impl UdfPointer for Arc<dyn AggregateUDFImpl + '_> {
 
 impl UdfPointer for Arc<dyn WindowUDFImpl + '_> {
     fn equals(&self, other: &(dyn WindowUDFImpl + '_)) -> bool {
-        self.as_ref().dyn_eq(other.as_any())
+        self.as_ref().dyn_eq(other as &dyn Any)
     }
 
     fn hash_value(&self) -> u64 {