Use AggregateUDFImpl::downcast_ref and is() instead of as &dyn Any patterns

Replaces `(udaf.inner().as_ref() as &dyn Any).downcast_ref::<T>()` and
`.is::<T>()` call sites with the new inherent methods on
`dyn AggregateUDFImpl`. Removes now-unused `std::any::Any` imports.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: timsaucer

datafusion/ffi/src/proto/logical_extension_codec.rs

@@ -680,7 +680,7 @@ mod tests {
 
         let returned_udf = foreign_codec.try_decode_udaf(udf.name(), &bytes)?;
 
-        assert!((returned_udf.inner().as_ref() as &dyn Any).is::<Sum>());
+        assert!(returned_udf.inner().is::<Sum>());
 
         Ok(())
     }

datafusion/ffi/src/proto/physical_extension_codec.rs

@@ -520,7 +520,7 @@ pub(crate) mod tests {
             buf.push(Self::MAGIC_NUMBER);
 
             let udf = node.inner();
-            let Some(_udf) = (udf.as_ref() as &dyn Any).downcast_ref::<Sum>() else {
+            let Some(_udf) = udf.downcast_ref::<Sum>() else {
                 return exec_err!("TestExtensionCodec only expects Sum UDAF");
             };
 
@@ -630,7 +630,7 @@ pub(crate) mod tests {
 
         let returned_udf = foreign_codec.try_decode_udaf(udf.name(), &bytes)?;
 
-        assert!((returned_udf.inner().as_ref() as &dyn Any).is::<Sum>());
+        assert!(returned_udf.inner().is::<Sum>());
 
         Ok(())
     }

datafusion/ffi/src/udaf/mod.rs

@@ -15,7 +15,6 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::any::Any;
 use std::ffi::c_void;
 use std::hash::{Hash, Hasher};
 use std::sync::Arc;
@@ -372,9 +371,7 @@ impl Clone for FFI_AggregateUDF {
 
 impl From<Arc<AggregateUDF>> for FFI_AggregateUDF {
     fn from(udaf: Arc<AggregateUDF>) -> Self {
-        if let Some(udaf) =
-            (udaf.inner().as_ref() as &dyn Any).downcast_ref::<ForeignAggregateUDF>()
-        {
+        if let Some(udaf) = udaf.inner().downcast_ref::<ForeignAggregateUDF>() {
             return udaf.udaf.clone();
         }
 
@@ -644,7 +641,6 @@ impl From<AggregateOrderSensitivity> for FFI_AggregateOrderSensitivity {
 
 #[cfg(test)]
 mod tests {
-    use std::any::Any;
     use std::collections::HashMap;
 
     use arrow::datatypes::Schema;
@@ -855,20 +851,12 @@ mod tests {
 
         // Verify local libraries can be downcast to their original
         let foreign_udaf: Arc<dyn AggregateUDFImpl> = (&ffi_udaf).into();
-        assert!(
-            (foreign_udaf.as_ref() as &dyn Any)
-                .downcast_ref::<Sum>()
-                .is_some()
-        );
+        assert!(foreign_udaf.is::<Sum>());
 
         // Verify different library markers generate foreign providers
         ffi_udaf.library_marker_id = crate::mock_foreign_marker_id;
         let foreign_udaf: Arc<dyn AggregateUDFImpl> = (&ffi_udaf).into();
-        assert!(
-            (foreign_udaf.as_ref() as &dyn Any)
-                .downcast_ref::<ForeignAggregateUDF>()
-                .is_some()
-        );
+        assert!(foreign_udaf.is::<ForeignAggregateUDF>());
 
         Ok(())
     }

datafusion/proto/tests/cases/roundtrip_physical_plan.rs

@@ -1306,8 +1306,7 @@ impl PhysicalExtensionCodec for UDFExtensionCodec {
 
     fn try_encode_udaf(&self, node: &AggregateUDF, buf: &mut Vec<u8>) -> Result<()> {
         let binding = node.inner();
-        if let Some(udf) = (binding.as_ref() as &dyn Any).downcast_ref::<MyAggregateUDF>()
-        {
+        if let Some(udf) = binding.downcast_ref::<MyAggregateUDF>() {
             let proto = MyAggregateUdfNode {
                 result: udf.result.clone(),
             };