Remove as_any on the PhysicalExpr trait (apache#21573)

## Which issue does this PR close?

- Addresses part of apache#21572

## Rationale for this change

This PR reduces the amount of boilerplate code that users need to write
for Physical Expressions.

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

Author: timsaucer

datafusion-examples/examples/custom_data_source/custom_file_casts.rs

@@ -185,7 +185,7 @@ impl PhysicalExprAdapter for CustomCastsPhysicalExprAdapter {
         // kernel / expression. For example, DataFusion Comet has a custom cast
         // kernel in its native Spark expression implementation.
         expr.transform(|expr| {
-            if let Some(cast) = expr.as_any().downcast_ref::<CastExpr>() {
+            if let Some(cast) = expr.downcast_ref::<CastExpr>() {
                 let input_data_type =
                     cast.expr().data_type(&self.physical_file_schema)?;
                 let output_data_type = cast.target_field().data_type();

datafusion-examples/examples/data_io/json_shredding.rs

@@ -311,20 +311,16 @@ impl ShreddedJsonRewriter {
         expr: Arc<dyn PhysicalExpr>,
         physical_file_schema: &Schema,
     ) -> Result<Transformed<Arc<dyn PhysicalExpr>>> {
-        if let Some(func) = expr.as_any().downcast_ref::<ScalarFunctionExpr>()
+        if let Some(func) = expr.downcast_ref::<ScalarFunctionExpr>()
             && func.name() == "json_get_str"
             && func.args().len() == 2
         {
             // Get the key from the first argument
-            if let Some(literal) = func.args()[0]
-                .as_any()
-                .downcast_ref::<expressions::Literal>()
+            if let Some(literal) = func.args()[0].downcast_ref::<expressions::Literal>()
                 && let ScalarValue::Utf8(Some(field_name)) = literal.value()
             {
                 // Get the column from the second argument
-                if let Some(column) = func.args()[1]
-                    .as_any()
-                    .downcast_ref::<expressions::Column>()
+                if let Some(column) = func.args()[1].downcast_ref::<expressions::Column>()
                 {
                     let column_name = column.name();
                     // Check if there's a flat column with underscore prefix

datafusion-examples/examples/proto/expression_deduplication.rs

@@ -129,7 +129,6 @@ pub async fn expression_deduplication() -> Result<()> {
     };
     let predicate = Arc::clone(filter_exec.predicate());
     let binary_expr = predicate
-        .as_any()
         .downcast_ref::<BinaryExpr>()
         .expect("Predicate is not a BinaryExpr");
     let left = &binary_expr.left();

datafusion/core/src/datasource/mod.rs

@@ -167,7 +167,7 @@ mod tests {
     impl PhysicalExprAdapter for TestPhysicalExprAdapter {
         fn rewrite(&self, expr: Arc<dyn PhysicalExpr>) -> Result<Arc<dyn PhysicalExpr>> {
             expr.transform(|e| {
-                if let Some(column) = e.as_any().downcast_ref::<Column>() {
+                if let Some(column) = e.downcast_ref::<Column>() {
                     // If column is "extra_column" and missing from physical schema, inject "foo"
                     if column.name() == "extra_column"
                         && self.physical_file_schema.index_of("extra_column").is_err()

datafusion/core/tests/fuzz_cases/equivalence/utils.rs

@@ -49,7 +49,7 @@ pub fn output_schema(
         let data_type = source.data_type(input_schema)?;
         let nullable = source.nullable(input_schema)?;
         for (target, _) in targets.iter() {
-            let Some(column) = target.as_any().downcast_ref::<Column>() else {
+            let Some(column) = target.downcast_ref::<Column>() else {
                 return plan_err!("Expects to have column");
             };
             fields.push(Field::new(column.name(), data_type.clone(), nullable));
@@ -282,7 +282,7 @@ fn get_representative_arr(
     schema: SchemaRef,
 ) -> Option<ArrayRef> {
     for expr in eq_group.iter() {
-        let col = expr.as_any().downcast_ref::<Column>().unwrap();
+        let col = expr.downcast_ref::<Column>().unwrap();
         let (idx, _field) = schema.column_with_name(col.name()).unwrap();
         if let Some(res) = &existing_vec[idx] {
             return Some(Arc::clone(res));
@@ -370,7 +370,7 @@ pub fn generate_table_for_eq_properties(
 
     // Fill constant columns
     for constant in eq_properties.constants() {
-        let col = constant.expr.as_any().downcast_ref::<Column>().unwrap();
+        let col = constant.expr.downcast_ref::<Column>().unwrap();
         let (idx, _field) = schema.column_with_name(col.name()).unwrap();
         let arr =
             Arc::new(Float64Array::from_iter_values(vec![0 as f64; n_elem])) as ArrayRef;
@@ -382,7 +382,7 @@ pub fn generate_table_for_eq_properties(
         let (sort_columns, indices): (Vec<_>, Vec<_>) = ordering
             .iter()
             .map(|PhysicalSortExpr { expr, options }| {
-                let col = expr.as_any().downcast_ref::<Column>().unwrap();
+                let col = expr.downcast_ref::<Column>().unwrap();
                 let (idx, _field) = schema.column_with_name(col.name()).unwrap();
                 let arr = generate_random_array(n_elem, n_distinct);
                 (
@@ -408,7 +408,7 @@ pub fn generate_table_for_eq_properties(
                 .unwrap_or_else(|| generate_random_array(n_elem, n_distinct));
 
         for expr in eq_group.iter() {
-            let col = expr.as_any().downcast_ref::<Column>().unwrap();
+            let col = expr.downcast_ref::<Column>().unwrap();
             let (idx, _field) = schema.column_with_name(col.name()).unwrap();
             schema_vec[idx] = Some(Arc::clone(&representative_array));
         }

datafusion/core/tests/fuzz_cases/sort_preserving_repartition_fuzz.rs

@@ -118,7 +118,7 @@ mod sp_repartition_fuzz_tests {
         schema: SchemaRef,
     ) -> Option<ArrayRef> {
         for expr in eq_group.iter() {
-            let col = expr.as_any().downcast_ref::<Column>().unwrap();
+            let col = expr.downcast_ref::<Column>().unwrap();
             let (idx, _field) = schema.column_with_name(col.name()).unwrap();
             if let Some(res) = &existing_vec[idx] {
                 return Some(res.clone());
@@ -149,7 +149,7 @@ mod sp_repartition_fuzz_tests {
 
         // Fill constant columns
         for constant in eq_properties.constants() {
-            let col = constant.expr.as_any().downcast_ref::<Column>().unwrap();
+            let col = constant.expr.downcast_ref::<Column>().unwrap();
             let (idx, _field) = schema.column_with_name(col.name()).unwrap();
             let arr =
                 Arc::new(UInt64Array::from_iter_values(vec![0; n_elem])) as ArrayRef;
@@ -161,7 +161,7 @@ mod sp_repartition_fuzz_tests {
             let (sort_columns, indices): (Vec<_>, Vec<_>) = ordering
                 .iter()
                 .map(|PhysicalSortExpr { expr, options }| {
-                    let col = expr.as_any().downcast_ref::<Column>().unwrap();
+                    let col = expr.downcast_ref::<Column>().unwrap();
                     let (idx, _field) = schema.column_with_name(col.name()).unwrap();
                     let arr = generate_random_array(n_elem, n_distinct);
                     (
@@ -187,7 +187,7 @@ mod sp_repartition_fuzz_tests {
                     .unwrap_or_else(|| generate_random_array(n_elem, n_distinct));
 
             for expr in eq_group.iter() {
-                let col = expr.as_any().downcast_ref::<Column>().unwrap();
+                let col = expr.downcast_ref::<Column>().unwrap();
                 let (idx, _field) = schema.column_with_name(col.name()).unwrap();
                 schema_vec[idx] = Some(representative_array.clone());
             }

datafusion/core/tests/parquet/expr_adapter.rs

@@ -490,7 +490,7 @@ impl PhysicalExprAdapter for CustomPhysicalExprAdapter {
     fn rewrite(&self, mut expr: Arc<dyn PhysicalExpr>) -> Result<Arc<dyn PhysicalExpr>> {
         expr = expr
             .transform(|expr| {
-                if let Some(column) = expr.as_any().downcast_ref::<Column>() {
+                if let Some(column) = expr.downcast_ref::<Column>() {
                     let field_name = column.name();
                     if self
                         .physical_file_schema

datafusion/core/tests/physical_optimizer/filter_pushdown.rs

@@ -2959,7 +2959,7 @@ async fn test_discover_dynamic_filters_via_expressions_api() {
 
         // Check expressions from this node using apply_expressions
         let _ = plan.apply_expressions(&mut |expr| {
-            if let Some(_df) = expr.as_any().downcast_ref::<DynamicFilterPhysicalExpr>() {
+            if let Some(_df) = expr.downcast_ref::<DynamicFilterPhysicalExpr>() {
                 count += 1;
             }
             Ok(TreeNodeRecursion::Continue)

datafusion/core/tests/physical_optimizer/join_selection.rs

@@ -759,7 +759,6 @@ async fn test_hash_join_swap_on_joins_with_projections(
 
 fn assert_col_expr(expr: &Arc<dyn PhysicalExpr>, name: &str, index: usize) {
     let col = expr
-        .as_any()
         .downcast_ref::<Column>()
         .expect("Projection items should be Column expression");
     assert_eq!(col.name(), name);

datafusion/core/tests/physical_optimizer/replace_with_order_preserving_variants.rs

@@ -1127,8 +1127,8 @@ fn hash_join_exec(
 ) -> Arc<dyn ExecutionPlan> {
     let left_on = col("c", &left.schema()).unwrap();
     let right_on = col("c", &right.schema()).unwrap();
-    let left_col = left_on.as_any().downcast_ref::<Column>().unwrap();
-    let right_col = right_on.as_any().downcast_ref::<Column>().unwrap();
+    let left_col = left_on.downcast_ref::<Column>().unwrap();
+    let right_col = right_on.downcast_ref::<Column>().unwrap();
     Arc::new(
         HashJoinExec::try_new(
             left,