chore: remove as_any from ExecutionPlan (#21263)

## Which issue does this PR close?

This is a follow on to #20812 and #21209 but treats `ExecutionPlan`.

## Rationale for this change

This PR reduces the amount of boilerplate code that users need to write
for execution plans.

## 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.

I have also implemented functions `is` and `downcast_ref` on the trait
object for ExecutionPlan and applied this same pattern to the udf, udaf,
and udwf implementations. This allows for a clean downcasting and type
checking.

## 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: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: timsaucer

datafusion-examples/examples/custom_data_source/adapter_serialization.rs

@@ -317,7 +317,7 @@ impl PhysicalProtoConverterExtension for AdapterPreservingCodec {
         extension_codec: &dyn PhysicalExtensionCodec,
     ) -> Result<PhysicalPlanNode> {
         // Check if this is a DataSourceExec with adapter
-        if let Some(exec) = plan.as_any().downcast_ref::<DataSourceExec>()
+        if let Some(exec) = plan.downcast_ref::<DataSourceExec>()
             && let Some(config) =
                 exec.data_source().as_any().downcast_ref::<FileScanConfig>()
             && let Some(adapter_factory) = &config.expr_adapter_factory
@@ -481,7 +481,7 @@ fn inject_adapter_into_plan(
     plan: Arc<dyn ExecutionPlan>,
     adapter_factory: Arc<dyn PhysicalExprAdapterFactory>,
 ) -> Result<Arc<dyn ExecutionPlan>> {
-    if let Some(exec) = plan.as_any().downcast_ref::<DataSourceExec>()
+    if let Some(exec) = plan.downcast_ref::<DataSourceExec>()
         && let Some(config) = exec.data_source().as_any().downcast_ref::<FileScanConfig>()
     {
         let new_config = FileScanConfigBuilder::from(config.clone())
@@ -497,7 +497,7 @@ fn inject_adapter_into_plan(
 fn verify_adapter_in_plan(plan: &Arc<dyn ExecutionPlan>, label: &str) -> bool {
     // Walk the plan tree to find DataSourceExec with adapter
     fn check_plan(plan: &dyn ExecutionPlan) -> bool {
-        if let Some(exec) = plan.as_any().downcast_ref::<DataSourceExec>()
+        if let Some(exec) = plan.downcast_ref::<DataSourceExec>()
             && let Some(config) =
                 exec.data_source().as_any().downcast_ref::<FileScanConfig>()
             && config.expr_adapter_factory.is_some()

datafusion-examples/examples/custom_data_source/custom_datasource.rs

@@ -235,10 +235,6 @@ impl ExecutionPlan for CustomExec {
         "CustomExec"
     }
 
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn properties(&self) -> &Arc<PlanProperties> {
         &self.cache
     }

datafusion-examples/examples/data_io/parquet_exec_visitor.rs

@@ -104,7 +104,7 @@ impl ExecutionPlanVisitor for ParquetExecVisitor {
     /// or `post_visit` (visit each node after its children/inputs)
     fn pre_visit(&mut self, plan: &dyn ExecutionPlan) -> Result<bool, Self::Error> {
         // If needed match on a specific `ExecutionPlan` node type
-        if let Some(data_source_exec) = plan.as_any().downcast_ref::<DataSourceExec>()
+        if let Some(data_source_exec) = plan.downcast_ref::<DataSourceExec>()
             && let Some((file_config, _)) =
                 data_source_exec.downcast_to_file_source::<ParquetSource>()
         {

datafusion-examples/examples/execution_monitoring/memory_pool_execution_plan.rs

@@ -43,7 +43,6 @@ use datafusion::physical_plan::{
 };
 use datafusion::prelude::*;
 use futures::stream::{StreamExt, TryStreamExt};
-use std::any::Any;
 use std::fmt;
 use std::sync::Arc;
 
@@ -226,10 +225,6 @@ impl ExecutionPlan for BufferingExecutionPlan {
         "BufferingExecutionPlan"
     }
 
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn schema(&self) -> SchemaRef {
         self.schema.clone()
     }

datafusion-examples/examples/proto/composed_extension_codec.rs

@@ -32,7 +32,6 @@
 //!           DeltaScan
 //! ```
 
-use std::any::Any;
 use std::fmt::Debug;
 use std::sync::Arc;
 
@@ -103,10 +102,6 @@ impl ExecutionPlan for ParentExec {
         "ParentExec"
     }
 
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn properties(&self) -> &Arc<datafusion::physical_plan::PlanProperties> {
         unreachable!()
     }
@@ -161,7 +156,7 @@ impl PhysicalExtensionCodec for ParentPhysicalExtensionCodec {
     }
 
     fn try_encode(&self, node: Arc<dyn ExecutionPlan>, buf: &mut Vec<u8>) -> Result<()> {
-        if node.as_any().downcast_ref::<ParentExec>().is_some() {
+        if node.is::<ParentExec>() {
             buf.extend_from_slice("ParentExec".as_bytes());
             Ok(())
         } else {
@@ -188,10 +183,6 @@ impl ExecutionPlan for ChildExec {
         "ChildExec"
     }
 
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn properties(&self) -> &Arc<datafusion::physical_plan::PlanProperties> {
         unreachable!()
     }
@@ -244,7 +235,7 @@ impl PhysicalExtensionCodec for ChildPhysicalExtensionCodec {
     }
 
     fn try_encode(&self, node: Arc<dyn ExecutionPlan>, buf: &mut Vec<u8>) -> Result<()> {
-        if node.as_any().downcast_ref::<ChildExec>().is_some() {
+        if node.is::<ChildExec>() {
             buf.extend_from_slice("ChildExec".as_bytes());
             Ok(())
         } else {

datafusion-examples/examples/proto/expression_deduplication.rs

@@ -124,8 +124,7 @@ pub async fn expression_deduplication() -> Result<()> {
 
     // Step 5: check that we deduplicated expressions
     println!("Step 5: Checking for deduplicated expressions...");
-    let Some(filter_exec) = deserialized_plan.as_any().downcast_ref::<FilterExec>()
-    else {
+    let Some(filter_exec) = deserialized_plan.downcast_ref::<FilterExec>() else {
         panic!("Deserialized plan is not a FilterExec");
     };
     let predicate = Arc::clone(filter_exec.predicate());

datafusion-examples/examples/relation_planner/table_sample.rs

@@ -80,7 +80,6 @@
 //! ```
 
 use std::{
-    any::Any,
     fmt::{self, Debug, Formatter},
     hash::{Hash, Hasher},
     pin::Pin,
@@ -682,10 +681,6 @@ impl ExecutionPlan for SampleExec {
         "SampleExec"
     }
 
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn properties(&self) -> &Arc<PlanProperties> {
         &self.cache
     }

datafusion/catalog/src/memory/table.rs

@@ -594,10 +594,6 @@ impl ExecutionPlan for DmlResultExec {
         "DmlResultExec"
     }
 
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn schema(&self) -> SchemaRef {
         Arc::clone(&self.schema)
     }

datafusion/core/src/datasource/listing/table.rs

@@ -107,6 +107,7 @@ impl ListingTableConfigExt for ListingTableConfig {
 
 #[cfg(test)]
 mod tests {
+
     #[cfg(feature = "parquet")]
     use crate::datasource::file_format::parquet::ParquetFormat;
     use crate::datasource::listing::table::ListingTableConfigExt;
@@ -404,7 +405,7 @@ mod tests {
             .await
             .expect("Empty execution plan");
 
-        assert!(scan.as_any().is::<EmptyExec>());
+        assert!(scan.is::<EmptyExec>());
         assert_eq!(
             columns(&scan.schema()),
             vec!["a".to_owned(), "p1".to_owned()]

datafusion/core/src/physical_planner.rs

@@ -3632,7 +3632,6 @@ mod tests {
 
         let execution_plan = plan(&logical_plan).await?;
         let final_hash_agg = execution_plan
-            .as_any()
             .downcast_ref::<AggregateExec>()
             .expect("hash aggregate");
         assert_eq!(
@@ -3660,7 +3659,6 @@ mod tests {
 
         let execution_plan = plan(&logical_plan).await?;
         let final_hash_agg = execution_plan
-            .as_any()
             .downcast_ref::<AggregateExec>()
             .expect("hash aggregate");
         assert_eq!(
@@ -3795,7 +3793,7 @@ mod tests {
             .unwrap();
 
         let plan = plan(&logical_plan).await.unwrap();
-        if let Some(plan) = plan.as_any().downcast_ref::<ExplainExec>() {
+        if let Some(plan) = plan.downcast_ref::<ExplainExec>() {
             let stringified_plans = plan.stringified_plans();
             assert!(stringified_plans.len() >= 4);
             assert!(
@@ -3863,7 +3861,7 @@ mod tests {
             .handle_explain(&explain, &ctx.state())
             .await
             .unwrap();
-        if let Some(plan) = plan.as_any().downcast_ref::<ExplainExec>() {
+        if let Some(plan) = plan.downcast_ref::<ExplainExec>() {
             let stringified_plans = plan.stringified_plans();
             assert_eq!(stringified_plans.len(), 1);
             assert_eq!(stringified_plans[0].plan.as_str(), "Test Err");
@@ -4003,10 +4001,6 @@ mod tests {
         }
 
         /// Return a reference to Any that can be used for downcasting
-        fn as_any(&self) -> &dyn Any {
-            self
-        }
-
         fn properties(&self) -> &Arc<PlanProperties> {
             &self.cache
         }
@@ -4169,9 +4163,6 @@ digraph {
         fn schema(&self) -> SchemaRef {
             Arc::new(Schema::empty())
         }
-        fn as_any(&self) -> &dyn Any {
-            unimplemented!()
-        }
         fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
             self.0.iter().collect::<Vec<_>>()
         }
@@ -4224,9 +4215,6 @@ digraph {
         ) -> Result<Arc<dyn ExecutionPlan>> {
             unimplemented!()
         }
-        fn as_any(&self) -> &dyn Any {
-            unimplemented!()
-        }
         fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
             unimplemented!()
         }
@@ -4351,9 +4339,6 @@ digraph {
         ) -> Result<Arc<dyn ExecutionPlan>> {
             unimplemented!()
         }
-        fn as_any(&self) -> &dyn Any {
-            unimplemented!()
-        }
         fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
             vec![]
         }
@@ -4765,6 +4750,6 @@ digraph {
             .unwrap();
 
         assert_eq!(plan.schema(), schema);
-        assert!(plan.as_any().is::<EmptyExec>());
+        assert!(plan.is::<EmptyExec>());
     }
 }