Remove trait function as_any from datafusion-datasource (#21576)

## Which issue does this PR close?

This addresses part of #21572

## Rationale for this change

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

## 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 traits 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/adapter_serialization.rs

@@ -318,8 +318,7 @@ impl PhysicalProtoConverterExtension for AdapterPreservingCodec {
     ) -> Result<PhysicalPlanNode> {
         // Check if this is a DataSourceExec with adapter
         if let Some(exec) = plan.downcast_ref::<DataSourceExec>()
-            && let Some(config) =
-                exec.data_source().as_any().downcast_ref::<FileScanConfig>()
+            && let Some(config) = exec.data_source().downcast_ref::<FileScanConfig>()
             && let Some(adapter_factory) = &config.expr_adapter_factory
             && let Some(tag) = extract_adapter_tag(adapter_factory.as_ref())
         {
@@ -482,7 +481,7 @@ fn inject_adapter_into_plan(
     adapter_factory: Arc<dyn PhysicalExprAdapterFactory>,
 ) -> Result<Arc<dyn ExecutionPlan>> {
     if let Some(exec) = plan.downcast_ref::<DataSourceExec>()
-        && let Some(config) = exec.data_source().as_any().downcast_ref::<FileScanConfig>()
+        && let Some(config) = exec.data_source().downcast_ref::<FileScanConfig>()
     {
         let new_config = FileScanConfigBuilder::from(config.clone())
             .with_expr_adapter(Some(adapter_factory))
@@ -498,8 +497,7 @@ 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.downcast_ref::<DataSourceExec>()
-            && let Some(config) =
-                exec.data_source().as_any().downcast_ref::<FileScanConfig>()
+            && let Some(config) = exec.data_source().downcast_ref::<FileScanConfig>()
             && config.expr_adapter_factory.is_some()
         {
             return true;

datafusion-examples/examples/custom_data_source/custom_file_format.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::{
     array::{AsArray, RecordBatch, StringArray, UInt8Array},
@@ -104,10 +104,6 @@ impl TSVFileFormat {
 
 #[async_trait::async_trait]
 impl FileFormat for TSVFileFormat {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn get_ext(&self) -> String {
         "tsv".to_string()
     }
@@ -207,10 +203,6 @@ impl FileFormatFactory for TSVFileFactory {
     fn default(&self) -> Arc<dyn FileFormat> {
         todo!()
     }
-
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
 }
 
 impl GetExt for TSVFileFactory {

datafusion-examples/examples/sql_ops/frontend.rs

@@ -31,7 +31,6 @@ use datafusion::optimizer::{
 use datafusion::sql::planner::{ContextProvider, SqlToRel};
 use datafusion::sql::sqlparser::dialect::PostgreSqlDialect;
 use datafusion::sql::sqlparser::parser::Parser;
-use std::any::Any;
 use std::sync::Arc;
 
 /// This example shows how to use DataFusion's SQL planner to parse SQL text and
@@ -190,10 +189,6 @@ struct MyTableSource {
 }
 
 impl TableSource for MyTableSource {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn schema(&self) -> SchemaRef {
         self.schema.clone()
     }

datafusion/catalog/src/default_table_source.rs

@@ -17,8 +17,8 @@
 
 //! Default TableSource implementation used in DataFusion physical plans
 
+use std::borrow::Cow;
 use std::sync::Arc;
-use std::{any::Any, borrow::Cow};
 
 use crate::TableProvider;
 
@@ -46,12 +46,6 @@ impl DefaultTableSource {
 }
 
 impl TableSource for DefaultTableSource {
-    /// Returns the table source as [`Any`] so that it can be
-    /// downcast to a specific implementation.
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     /// Get a reference to the schema for this table
     fn schema(&self) -> SchemaRef {
         self.table_provider.schema()
@@ -97,11 +91,7 @@ pub fn provider_as_source(
 pub fn source_as_provider(
     source: &Arc<dyn TableSource>,
 ) -> datafusion_common::Result<Arc<dyn TableProvider>> {
-    match source
-        .as_ref()
-        .as_any()
-        .downcast_ref::<DefaultTableSource>()
-    {
+    match source.as_ref().downcast_ref::<DefaultTableSource>() {
         Some(source) => Ok(Arc::clone(&source.table_provider)),
         _ => internal_err!("TableSource was not DefaultTableSource"),
     }

datafusion/catalog/src/stream.rs

@@ -17,7 +17,6 @@
 
 //! TableProvider for stream sources, such as FIFO files
 
-use std::any::Any;
 use std::fmt::Formatter;
 use std::fs::{File, OpenOptions};
 use std::io::BufReader;
@@ -401,10 +400,6 @@ impl DisplayAs for StreamWrite {
 
 #[async_trait]
 impl DataSink for StreamWrite {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn schema(&self) -> &SchemaRef {
         self.0.source.schema()
     }

datafusion/core/src/datasource/listing_table_factory.rs

@@ -297,7 +297,7 @@ mod tests {
         let listing_table = table_provider.downcast_ref::<ListingTable>().unwrap();
 
         let format = listing_table.options().format.clone();
-        let csv_format = format.as_any().downcast_ref::<CsvFormat>().unwrap();
+        let csv_format = format.downcast_ref::<CsvFormat>().unwrap();
         let csv_options = csv_format.options().clone();
         assert_eq!(csv_options.schema_infer_max_rec, Some(1000));
         let listing_options = listing_table.options();
@@ -332,7 +332,7 @@ mod tests {
 
         // Verify compression is used
         let format = listing_table.options().format.clone();
-        let csv_format = format.as_any().downcast_ref::<CsvFormat>().unwrap();
+        let csv_format = format.downcast_ref::<CsvFormat>().unwrap();
         let csv_options = csv_format.options().clone();
         assert_eq!(csv_options.compression, CompressionTypeVariant::GZIP);
 

datafusion/core/src/physical_planner.rs

@@ -205,7 +205,7 @@ pub trait ExtensionPlanner {
     ///         _session_state: &SessionState,
     ///     ) -> Result<Option<Arc<dyn ExecutionPlan>>> {
     ///         // Check if this is your custom table source
-    ///         if scan.source.as_any().is::<MyCustomTableSource>() {
+    ///         if scan.source.is::<MyCustomTableSource>() {
     ///             // Create a custom execution plan for your table source
     ///             let exec = MyCustomExec::new(
     ///                 scan.table_name.clone(),
@@ -732,9 +732,7 @@ impl DefaultPhysicalPlanner {
                 op: WriteOp::Insert(insert_op),
                 ..
             }) => {
-                if let Some(provider) =
-                    target.as_any().downcast_ref::<DefaultTableSource>()
-                {
+                if let Some(provider) = target.downcast_ref::<DefaultTableSource>() {
                     let input_exec = children.one()?;
                     provider
                         .table_provider
@@ -753,9 +751,7 @@ impl DefaultPhysicalPlanner {
                 input,
                 ..
             }) => {
-                if let Some(provider) =
-                    target.as_any().downcast_ref::<DefaultTableSource>()
-                {
+                if let Some(provider) = target.downcast_ref::<DefaultTableSource>() {
                     let filters = extract_dml_filters(input, table_name)?;
                     provider
                         .table_provider
@@ -777,9 +773,7 @@ impl DefaultPhysicalPlanner {
                 input,
                 ..
             }) => {
-                if let Some(provider) =
-                    target.as_any().downcast_ref::<DefaultTableSource>()
-                {
+                if let Some(provider) = target.downcast_ref::<DefaultTableSource>() {
                     // For UPDATE, the assignments are encoded in the projection of input
                     // We pass the filters and let the provider handle the projection
                     let filters = extract_dml_filters(input, table_name)?;
@@ -804,9 +798,7 @@ impl DefaultPhysicalPlanner {
                 op: WriteOp::Truncate,
                 ..
             }) => {
-                if let Some(provider) =
-                    target.as_any().downcast_ref::<DefaultTableSource>()
-                {
+                if let Some(provider) = target.downcast_ref::<DefaultTableSource>() {
                     provider
                         .table_provider
                         .truncate(session_state)
@@ -3112,7 +3104,6 @@ impl<'n> TreeNodeVisitor<'n> for InvariantChecker {
 
 #[cfg(test)]
 mod tests {
-    use std::any::Any;
     use std::cmp::Ordering;
     use std::fmt::{self, Debug};
     use std::ops::{BitAnd, Not};
@@ -4680,10 +4671,6 @@ digraph {
     }
 
     impl TableSource for MockTableSource {
-        fn as_any(&self) -> &dyn Any {
-            self
-        }
-
         fn schema(&self) -> SchemaRef {
             Arc::clone(&self.schema)
         }
@@ -4710,7 +4697,7 @@ digraph {
             scan: &TableScan,
             _session_state: &SessionState,
         ) -> Result<Option<Arc<dyn ExecutionPlan>>> {
-            if scan.source.as_any().is::<MockTableSource>() {
+            if scan.source.is::<MockTableSource>() {
                 Ok(Some(Arc::new(EmptyExec::new(Arc::clone(
                     scan.projected_schema.inner(),
                 )))))

datafusion/core/tests/optimizer/mod.rs

@@ -19,7 +19,6 @@
 //! datafusion-functions crate.
 
 use insta::assert_snapshot;
-use std::any::Any;
 use std::collections::HashMap;
 use std::sync::Arc;
 
@@ -251,10 +250,6 @@ struct MyTableSource {
 }
 
 impl TableSource for MyTableSource {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn schema(&self) -> SchemaRef {
         self.schema.clone()
     }

datafusion/core/tests/parquet/file_statistics.rs

@@ -198,10 +198,7 @@ async fn list_files_with_session_level_cache() {
     let exec1 = table1.scan(&state1, None, &[], None).await.unwrap();
     let data_source_exec = exec1.downcast_ref::<DataSourceExec>().unwrap();
     let data_source = data_source_exec.data_source();
-    let parquet1 = data_source
-        .as_any()
-        .downcast_ref::<FileScanConfig>()
-        .unwrap();
+    let parquet1 = data_source.downcast_ref::<FileScanConfig>().unwrap();
 
     assert_eq!(get_list_file_cache_size(&state1), 1);
     let fg = &parquet1.file_groups;
@@ -214,10 +211,7 @@ async fn list_files_with_session_level_cache() {
     let exec2 = table2.scan(&state2, None, &[], None).await.unwrap();
     let data_source_exec = exec2.downcast_ref::<DataSourceExec>().unwrap();
     let data_source = data_source_exec.data_source();
-    let parquet2 = data_source
-        .as_any()
-        .downcast_ref::<FileScanConfig>()
-        .unwrap();
+    let parquet2 = data_source.downcast_ref::<FileScanConfig>().unwrap();
 
     assert_eq!(get_list_file_cache_size(&state2), 1);
     let fg2 = &parquet2.file_groups;
@@ -230,10 +224,7 @@ async fn list_files_with_session_level_cache() {
     let exec3 = table1.scan(&state1, None, &[], None).await.unwrap();
     let data_source_exec = exec3.downcast_ref::<DataSourceExec>().unwrap();
     let data_source = data_source_exec.data_source();
-    let parquet3 = data_source
-        .as_any()
-        .downcast_ref::<FileScanConfig>()
-        .unwrap();
+    let parquet3 = data_source.downcast_ref::<FileScanConfig>().unwrap();
 
     assert_eq!(get_list_file_cache_size(&state1), 1);
     let fg = &parquet3.file_groups;

datafusion/core/tests/physical_optimizer/projection_pushdown.rs

@@ -518,7 +518,6 @@ fn test_memory_after_projection() -> Result<()> {
             .downcast_ref::<DataSourceExec>()
             .unwrap()
             .data_source()
-            .as_any()
             .downcast_ref::<MemorySourceConfig>()
             .unwrap()
             .projection()