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

Replaces remaining `(x.as_ref() as &dyn Any).downcast_ref::<T>()` and
`(x as &dyn std::any::Any).downcast_ref::<T>()` patterns with the new
inherent methods on `dyn ExecutionPlan`. Also cleans up redundant
`.as_ref()` calls before `.downcast_ref()` / `.is()` on Arc values,
and simplifies `.downcast_ref::<T>().is_some()` to `.is::<T>()` where
applicable. Removes now-unused `std::any::Any` imports.

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

Author: timsaucer

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;
 
@@ -157,10 +156,7 @@ impl PhysicalExtensionCodec for ParentPhysicalExtensionCodec {
     }
 
     fn try_encode(&self, node: Arc<dyn ExecutionPlan>, buf: &mut Vec<u8>) -> Result<()> {
-        if (node.as_ref() as &dyn Any)
-            .downcast_ref::<ParentExec>()
-            .is_some()
-        {
+        if node.is::<ParentExec>() {
             buf.extend_from_slice("ParentExec".as_bytes());
             Ok(())
         } else {
@@ -239,10 +235,7 @@ impl PhysicalExtensionCodec for ChildPhysicalExtensionCodec {
     }
 
     fn try_encode(&self, node: Arc<dyn ExecutionPlan>, buf: &mut Vec<u8>) -> Result<()> {
-        if (node.as_ref() as &dyn Any)
-            .downcast_ref::<ChildExec>()
-            .is_some()
-        {
+        if node.is::<ChildExec>() {
             buf.extend_from_slice("ChildExec".as_bytes());
             Ok(())
         } else {

datafusion/core/src/physical_planner.rs

@@ -3624,7 +3624,7 @@ mod tests {
             .build()?;
 
         let execution_plan = plan(&logical_plan).await?;
-        let final_hash_agg = (execution_plan.as_ref() as &dyn Any)
+        let final_hash_agg = execution_plan
             .downcast_ref::<AggregateExec>()
             .expect("hash aggregate");
         assert_eq!(
@@ -3651,7 +3651,7 @@ mod tests {
             .build()?;
 
         let execution_plan = plan(&logical_plan).await?;
-        let final_hash_agg = (execution_plan.as_ref() as &dyn Any)
+        let final_hash_agg = execution_plan
             .downcast_ref::<AggregateExec>()
             .expect("hash aggregate");
         assert_eq!(

datafusion/core/src/test_util/parquet.rs

@@ -196,8 +196,7 @@ impl TestParquetFile {
     /// Recursively searches for DataSourceExec and returns the metrics
     /// on the first one it finds
     pub fn parquet_metrics(plan: &Arc<dyn ExecutionPlan>) -> Option<MetricsSet> {
-        if let Some(data_source_exec) =
-            (plan.as_ref() as &dyn std::any::Any).downcast_ref::<DataSourceExec>()
+        if let Some(data_source_exec) = plan.downcast_ref::<DataSourceExec>()
             && data_source_exec
                 .downcast_to_file_source::<ParquetSource>()
                 .is_some()

datafusion/core/tests/fuzz_cases/aggregate_fuzz.rs

@@ -547,9 +547,7 @@ async fn verify_ordered_aggregate(frame: &DataFrame, expected_sort: bool) {
         type Node = Arc<dyn ExecutionPlan>;
 
         fn f_down(&mut self, node: &'n Self::Node) -> Result<TreeNodeRecursion> {
-            if let Some(exec) =
-                (node.as_ref() as &dyn std::any::Any).downcast_ref::<AggregateExec>()
-            {
+            if let Some(exec) = node.downcast_ref::<AggregateExec>() {
                 if self.expected_sort {
                     assert!(matches!(
                         exec.input_order_mode(),

datafusion/core/tests/parquet/file_statistics.rs

@@ -15,7 +15,6 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::any::Any;
 use std::fs;
 use std::sync::Arc;
 
@@ -197,9 +196,7 @@ async fn list_files_with_session_level_cache() {
     //Session 1 first time list files
     assert_eq!(get_list_file_cache_size(&state1), 0);
     let exec1 = table1.scan(&state1, None, &[], None).await.unwrap();
-    let data_source_exec = (exec1.as_ref() as &dyn Any)
-        .downcast_ref::<DataSourceExec>()
-        .unwrap();
+    let data_source_exec = exec1.downcast_ref::<DataSourceExec>().unwrap();
     let data_source = data_source_exec.data_source();
     let parquet1 = data_source
         .as_any()
@@ -215,9 +212,7 @@ async fn list_files_with_session_level_cache() {
     //check session 1 cache result not show in session 2
     assert_eq!(get_list_file_cache_size(&state2), 0);
     let exec2 = table2.scan(&state2, None, &[], None).await.unwrap();
-    let data_source_exec = (exec2.as_ref() as &dyn Any)
-        .downcast_ref::<DataSourceExec>()
-        .unwrap();
+    let data_source_exec = exec2.downcast_ref::<DataSourceExec>().unwrap();
     let data_source = data_source_exec.data_source();
     let parquet2 = data_source
         .as_any()
@@ -233,9 +228,7 @@ async fn list_files_with_session_level_cache() {
     //check session 1 cache result not show in session 2
     assert_eq!(get_list_file_cache_size(&state1), 1);
     let exec3 = table1.scan(&state1, None, &[], None).await.unwrap();
-    let data_source_exec = (exec3.as_ref() as &dyn Any)
-        .downcast_ref::<DataSourceExec>()
-        .unwrap();
+    let data_source_exec = exec3.downcast_ref::<DataSourceExec>().unwrap();
     let data_source = data_source_exec.data_source();
     let parquet3 = data_source
         .as_any()

datafusion/core/tests/physical_optimizer/enforce_distribution.rs

@@ -15,7 +15,6 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::any::Any;
 use std::fmt::Debug;
 use std::ops::Deref;
 use std::sync::Arc;
@@ -3892,7 +3891,8 @@ fn test_replace_order_preserving_variants_with_fetch() -> Result<()> {
     let result = replace_order_preserving_variants(dist_context)?;
 
     // Verify the plan was transformed to CoalescePartitionsExec
-    (result.plan.as_ref() as &dyn Any)
+    result
+        .plan
         .downcast_ref::<CoalescePartitionsExec>()
         .expect("Expected CoalescePartitionsExec");
 

datafusion/core/tests/physical_optimizer/filter_pushdown.rs

@@ -15,7 +15,6 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::any::Any;
 use std::sync::{Arc, LazyLock};
 
 use arrow::{
@@ -4324,7 +4323,7 @@ async fn test_hashjoin_dynamic_filter_pushdown_is_used() {
             .unwrap();
 
         // Get the HashJoinExec to check the dynamic filter
-        let hash_join = (plan.as_ref() as &dyn Any)
+        let hash_join = plan
             .downcast_ref::<HashJoinExec>()
             .expect("Plan should be HashJoinExec");
 

datafusion/core/tests/physical_optimizer/join_selection.rs

@@ -232,7 +232,6 @@ async fn test_join_with_swap() {
         .unwrap();
 
     let swapping_projection = optimized_join
-        .as_ref()
         .downcast_ref::<ProjectionExec>()
         .expect("A proj is required to swap columns back to their original order");
 
@@ -246,7 +245,6 @@ async fn test_join_with_swap() {
 
     let swapped_join = swapping_projection
         .input()
-        .as_ref()
         .downcast_ref::<HashJoinExec>()
         .expect("The type of the plan should not be changed");
 
@@ -295,7 +293,6 @@ async fn test_left_join_no_swap() {
         .unwrap();
 
     let swapped_join = optimized_join
-        .as_ref()
         .downcast_ref::<HashJoinExec>()
         .expect("The type of the plan should not be changed");
 
@@ -345,12 +342,9 @@ async fn test_join_with_swap_semi() {
             .optimize(Arc::new(join), &ConfigOptions::new())
             .unwrap();
 
-        let swapped_join = optimized_join
-            .as_ref()
-            .downcast_ref::<HashJoinExec>()
-            .expect(
-                "A proj is not required to swap columns back to their original order",
-            );
+        let swapped_join = optimized_join.downcast_ref::<HashJoinExec>().expect(
+            "A proj is not required to swap columns back to their original order",
+        );
 
         assert_eq!(swapped_join.schema().fields().len(), 1);
         assert_eq!(
@@ -401,12 +395,9 @@ async fn test_join_with_swap_mark() {
             .optimize(Arc::new(join), &ConfigOptions::new())
             .unwrap();
 
-        let swapped_join = optimized_join
-            .as_ref()
-            .downcast_ref::<HashJoinExec>()
-            .expect(
-                "A proj is not required to swap columns back to their original order",
-            );
+        let swapped_join = optimized_join.downcast_ref::<HashJoinExec>().expect(
+            "A proj is not required to swap columns back to their original order",
+        );
 
         assert_eq!(swapped_join.schema().fields().len(), 2);
         assert_eq!(
@@ -534,7 +525,6 @@ async fn test_join_no_swap() {
         .unwrap();
 
     let swapped_join = optimized_join
-        .as_ref()
         .downcast_ref::<HashJoinExec>()
         .expect("The type of the plan should not be changed");
 
@@ -583,7 +573,6 @@ async fn test_nl_join_with_swap(join_type: JoinType) {
         .unwrap();
 
     let swapping_projection = optimized_join
-        .as_ref()
         .downcast_ref::<ProjectionExec>()
         .expect("A proj is required to swap columns back to their original order");
 
@@ -597,7 +586,6 @@ async fn test_nl_join_with_swap(join_type: JoinType) {
 
     let swapped_join = swapping_projection
         .input()
-        .as_ref()
         .downcast_ref::<NestedLoopJoinExec>()
         .expect("The type of the plan should not be changed");
 
@@ -664,7 +652,6 @@ async fn test_nl_join_with_swap_no_proj(join_type: JoinType) {
         .unwrap();
 
     let swapped_join = optimized_join
-        .as_ref()
         .downcast_ref::<NestedLoopJoinExec>()
         .expect("The type of the plan should not be changed");
 
@@ -759,7 +746,6 @@ async fn test_hash_join_swap_on_joins_with_projections(
         .swap_inputs(PartitionMode::Partitioned)
         .expect("swap_hash_join must support joins with projections");
     let swapped_join = swapped
-        .as_ref()
         .downcast_ref::<HashJoinExec>()
         .expect(
             "ProjectionExec won't be added above if HashJoinExec contains embedded projection",
@@ -928,18 +914,15 @@ fn check_join_partition_mode(
 
     if !is_swapped {
         let swapped_join = optimized_join
-            .as_ref()
             .downcast_ref::<HashJoinExec>()
             .expect("The type of the plan should not be changed");
         assert_eq!(*swapped_join.partition_mode(), expected_mode);
     } else {
         let swapping_projection = optimized_join
-            .as_ref()
             .downcast_ref::<ProjectionExec>()
             .expect("A proj is required to swap columns back to their original order");
         let swapped_join = swapping_projection
             .input()
-            .as_ref()
             .downcast_ref::<HashJoinExec>()
             .expect("The type of the plan should not be changed");
 
@@ -1589,10 +1572,9 @@ async fn test_join_with_maybe_swap_unbounded_case(t: TestCase) -> Result<()> {
         JoinSelection::new().optimize(Arc::clone(&join), &ConfigOptions::new())?;
 
     // If swap did happen
-    let projection_added = optimized_join_plan.as_ref().is::<ProjectionExec>();
+    let projection_added = optimized_join_plan.is::<ProjectionExec>();
     let plan = if projection_added {
         let proj = optimized_join_plan
-            .as_ref()
             .downcast_ref::<ProjectionExec>()
             .expect("A proj is required to swap columns back to their original order");
         Arc::<dyn ExecutionPlan>::clone(proj.input())
@@ -1606,7 +1588,7 @@ async fn test_join_with_maybe_swap_unbounded_case(t: TestCase) -> Result<()> {
         join_type,
         mode,
         ..
-    }) = plan.as_ref().downcast_ref::<HashJoinExec>()
+    }) = plan.downcast_ref::<HashJoinExec>()
     {
         let left_changed = Arc::ptr_eq(left, &right_exec);
         let right_changed = Arc::ptr_eq(right, &left_exec);

datafusion/core/tests/physical_optimizer/projection_pushdown.rs

@@ -15,7 +15,6 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::any::Any;
 use std::sync::Arc;
 
 use arrow::compute::SortOptions;
@@ -514,7 +513,8 @@ fn test_memory_after_projection() -> Result<()> {
     );
 
     assert_eq!(
-        (after_optimize.clone().as_ref() as &dyn Any)
+        after_optimize
+            .clone()
             .downcast_ref::<DataSourceExec>()
             .unwrap()
             .data_source()
@@ -597,9 +597,7 @@ fn test_streaming_table_after_projection() -> Result<()> {
     let after_optimize =
         ProjectionPushdown::new().optimize(projection, &ConfigOptions::new())?;
 
-    let result = (after_optimize.as_ref() as &dyn Any)
-        .downcast_ref::<StreamingTableExec>()
-        .unwrap();
+    let result = after_optimize.downcast_ref::<StreamingTableExec>().unwrap();
     assert_eq!(
         result.partition_schema(),
         &Arc::new(Schema::new(vec![
@@ -789,7 +787,7 @@ fn test_output_req_after_projection() -> Result<()> {
         .into(),
     );
     assert_eq!(
-        (after_optimize.as_ref() as &dyn Any)
+        after_optimize
             .downcast_ref::<OutputRequirementExec>()
             .unwrap()
             .required_input_ordering()[0]
@@ -801,7 +799,7 @@ fn test_output_req_after_projection() -> Result<()> {
         Arc::new(Column::new("new_a", 1)),
         Arc::new(Column::new("b", 2)),
     ];
-    if let Distribution::HashPartitioned(vec) = (after_optimize.as_ref() as &dyn Any)
+    if let Distribution::HashPartitioned(vec) = after_optimize
         .downcast_ref::<OutputRequirementExec>()
         .unwrap()
         .required_input_distribution()[0]
@@ -1031,7 +1029,7 @@ fn test_join_after_projection() -> Result<()> {
 
     assert_eq!(
         expected_filter_col_ind,
-        (after_optimize.as_ref() as &dyn Any)
+        after_optimize
             .downcast_ref::<SymmetricHashJoinExec>()
             .unwrap()
             .filter()
@@ -1394,7 +1392,7 @@ fn test_repartition_after_projection() -> Result<()> {
     );
 
     assert_eq!(
-        (after_optimize.as_ref() as &dyn Any)
+        after_optimize
             .downcast_ref::<RepartitionExec>()
             .unwrap()
             .partitioning()

datafusion/core/tests/user_defined/insert_operation.rs

@@ -58,9 +58,7 @@ async fn insert_operation_is_passed_correctly_to_table_provider() {
 async fn assert_insert_op(ctx: &SessionContext, sql: &str, insert_op: InsertOp) {
     let df = ctx.sql(sql).await.unwrap();
     let plan = df.create_physical_plan().await.unwrap();
-    let exec = (plan.as_ref() as &dyn Any)
-        .downcast_ref::<TestInsertExec>()
-        .unwrap();
+    let exec = plan.downcast_ref::<TestInsertExec>().unwrap();
     assert_eq!(exec.op, insert_op);
 }