You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/04/17 13:09:47 UTC
(datafusion) branch gh-readonly-queue/main/pr-21576-6cd7e8395820557f6b0d86125d2c8ea5c4b01b43 created (now 03b390db7b)
This is an automated email from the ASF dual-hosted git repository.
github-bot pushed a change to branch gh-readonly-queue/main/pr-21576-6cd7e8395820557f6b0d86125d2c8ea5c4b01b43
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 03b390db7b Remove trait function `as_any` from datafusion-datasource (#21576)
This branch includes the following new commits:
new 03b390db7b Remove trait function `as_any` from datafusion-datasource (#21576)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: Remove trait function `as_any` from datafusion-datasource (#21576)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-bot pushed a commit to branch gh-readonly-queue/main/pr-21576-6cd7e8395820557f6b0d86125d2c8ea5c4b01b43
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 03b390db7b5788476f9745eb9a4a4b5a873a4612
Author: Tim Saucer <ti...@gmail.com>
AuthorDate: Fri Apr 17 09:09:26 2026 -0400
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) <no...@anthropic.com>
---
.../custom_data_source/adapter_serialization.rs | 8 +++---
.../custom_data_source/custom_file_format.rs | 10 +-------
datafusion-examples/examples/sql_ops/frontend.rs | 5 ----
datafusion/catalog/src/default_table_source.rs | 14 ++--------
datafusion/catalog/src/stream.rs | 5 ----
.../core/src/datasource/listing_table_factory.rs | 4 +--
datafusion/core/src/physical_planner.rs | 25 +++++-------------
datafusion/core/tests/optimizer/mod.rs | 5 ----
datafusion/core/tests/parquet/file_statistics.rs | 15 +++--------
.../physical_optimizer/projection_pushdown.rs | 1 -
.../tests/physical_optimizer/pushdown_utils.rs | 5 ----
datafusion/datasource-arrow/src/file_format.rs | 15 ++---------
datafusion/datasource-arrow/src/source.rs | 6 +----
datafusion/datasource-avro/src/file_format.rs | 9 -------
datafusion/datasource-avro/src/source.rs | 5 ----
datafusion/datasource-csv/src/file_format.rs | 14 ----------
datafusion/datasource-csv/src/source.rs | 5 ----
datafusion/datasource-json/src/file_format.rs | 13 ----------
datafusion/datasource-json/src/source.rs | 5 ----
datafusion/datasource-parquet/src/file_format.rs | 14 ----------
datafusion/datasource-parquet/src/source.rs | 7 +----
datafusion/datasource/src/file.rs | 16 +++++++++---
datafusion/datasource/src/file_format.rs | 30 +++++++++++++++-------
datafusion/datasource/src/file_scan_config/mod.rs | 28 +-------------------
datafusion/datasource/src/memory.rs | 24 ++++-------------
datafusion/datasource/src/sink.rs | 18 +++++++++----
datafusion/datasource/src/source.rs | 23 ++++++++++++-----
datafusion/datasource/src/test_util.rs | 4 ---
datafusion/expr/src/logical_plan/builder.rs | 5 ----
datafusion/expr/src/table_source.rs | 14 +++++++---
datafusion/optimizer/src/push_down_filter.rs | 5 ----
.../optimizer/tests/optimizer_integration.rs | 9 -------
datafusion/proto/src/logical_plan/file_formats.rs | 16 +++++-------
datafusion/proto/src/logical_plan/mod.rs | 24 ++++++++---------
datafusion/proto/src/physical_plan/mod.rs | 26 +++++++++----------
.../proto/tests/cases/roundtrip_logical_plan.rs | 6 -----
.../proto/tests/cases/roundtrip_physical_plan.rs | 8 +-----
datafusion/sql/tests/common/mod.rs | 4 ---
docs/source/library-user-guide/upgrading/54.0.0.md | 2 +-
39 files changed, 132 insertions(+), 320 deletions(-)
diff --git a/datafusion-examples/examples/custom_data_source/adapter_serialization.rs b/datafusion-examples/examples/custom_data_source/adapter_serialization.rs
index df1dacf23a..22f8589d6f 100644
--- a/datafusion-examples/examples/custom_data_source/adapter_serialization.rs
+++ b/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;
diff --git a/datafusion-examples/examples/custom_data_source/custom_file_format.rs b/datafusion-examples/examples/custom_data_source/custom_file_format.rs
index 6817beec41..0cfbe11877 100644
--- a/datafusion-examples/examples/custom_data_source/custom_file_format.rs
+++ b/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 {
diff --git a/datafusion-examples/examples/sql_ops/frontend.rs b/datafusion-examples/examples/sql_ops/frontend.rs
index 025fe47e75..d7704f74db 100644
--- a/datafusion-examples/examples/sql_ops/frontend.rs
+++ b/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()
}
diff --git a/datafusion/catalog/src/default_table_source.rs b/datafusion/catalog/src/default_table_source.rs
index a06fcdbe0b..60f85891d6 100644
--- a/datafusion/catalog/src/default_table_source.rs
+++ b/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"),
}
diff --git a/datafusion/catalog/src/stream.rs b/datafusion/catalog/src/stream.rs
index c58ff939ee..8501ea6590 100644
--- a/datafusion/catalog/src/stream.rs
+++ b/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()
}
diff --git a/datafusion/core/src/datasource/listing_table_factory.rs b/datafusion/core/src/datasource/listing_table_factory.rs
index d1ee301d91..31d6e36657 100644
--- a/datafusion/core/src/datasource/listing_table_factory.rs
+++ b/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);
diff --git a/datafusion/core/src/physical_planner.rs b/datafusion/core/src/physical_planner.rs
index 33d02fc622..a025446aa3 100644
--- a/datafusion/core/src/physical_planner.rs
+++ b/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(),
)))))
diff --git a/datafusion/core/tests/optimizer/mod.rs b/datafusion/core/tests/optimizer/mod.rs
index 6466e9ad96..72d50760f1 100644
--- a/datafusion/core/tests/optimizer/mod.rs
+++ b/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()
}
diff --git a/datafusion/core/tests/parquet/file_statistics.rs b/datafusion/core/tests/parquet/file_statistics.rs
index c96df5c509..84396be8a6 100644
--- a/datafusion/core/tests/parquet/file_statistics.rs
+++ b/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;
diff --git a/datafusion/core/tests/physical_optimizer/projection_pushdown.rs b/datafusion/core/tests/physical_optimizer/projection_pushdown.rs
index 6018d714c5..6635220cf2 100644
--- a/datafusion/core/tests/physical_optimizer/projection_pushdown.rs
+++ b/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()
diff --git a/datafusion/core/tests/physical_optimizer/pushdown_utils.rs b/datafusion/core/tests/physical_optimizer/pushdown_utils.rs
index e8c02d4efc..8b659e757a 100644
--- a/datafusion/core/tests/physical_optimizer/pushdown_utils.rs
+++ b/datafusion/core/tests/physical_optimizer/pushdown_utils.rs
@@ -43,7 +43,6 @@ use futures::StreamExt;
use futures::{FutureExt, Stream};
use object_store::ObjectStore;
use std::{
- any::Any,
fmt::{Display, Formatter},
pin::Pin,
sync::Arc,
@@ -145,10 +144,6 @@ impl FileSource for TestSource {
self.predicate.clone()
}
- fn as_any(&self) -> &dyn Any {
- todo!("should not be called")
- }
-
fn with_batch_size(&self, batch_size: usize) -> Arc<dyn FileSource> {
Arc::new(TestSource {
batch_size: Some(batch_size),
diff --git a/datafusion/datasource-arrow/src/file_format.rs b/datafusion/datasource-arrow/src/file_format.rs
index 688ae1483b..fe5bba98ca 100644
--- a/datafusion/datasource-arrow/src/file_format.rs
+++ b/datafusion/datasource-arrow/src/file_format.rs
@@ -19,7 +19,6 @@
//!
//! Works with files following the [Arrow IPC format](https://arrow.apache.org/docs/format/Columnar.html#ipc-file-format)
-use std::any::Any;
use std::collections::HashMap;
use std::fmt::{self, Debug};
use std::io::{Seek, SeekFrom};
@@ -98,10 +97,6 @@ impl FileFormatFactory for ArrowFormatFactory {
fn default(&self) -> Arc<dyn FileFormat> {
Arc::new(ArrowFormat)
}
-
- fn as_any(&self) -> &dyn Any {
- self
- }
}
impl GetExt for ArrowFormatFactory {
@@ -117,10 +112,6 @@ pub struct ArrowFormat;
#[async_trait]
impl FileFormat for ArrowFormat {
- fn as_any(&self) -> &dyn Any {
- self
- }
-
fn get_ext(&self) -> String {
ArrowFormatFactory::new().get_ext()
}
@@ -374,10 +365,6 @@ impl DisplayAs for ArrowFileSink {
#[async_trait]
impl DataSink for ArrowFileSink {
- fn as_any(&self) -> &dyn Any {
- self
- }
-
fn schema(&self) -> &SchemaRef {
self.config.output_schema()
}
@@ -549,6 +536,8 @@ async fn is_object_in_arrow_ipc_file_format(
mod tests {
use super::*;
+ use std::any::Any;
+
use chrono::DateTime;
use datafusion_common::DFSchema;
use datafusion_common::config::TableOptions;
diff --git a/datafusion/datasource-arrow/src/source.rs b/datafusion/datasource-arrow/src/source.rs
index 0226dadd3d..061f130f24 100644
--- a/datafusion/datasource-arrow/src/source.rs
+++ b/datafusion/datasource-arrow/src/source.rs
@@ -31,8 +31,8 @@
//! refers to the Arrow IPC stream format, not streaming I/O. Both formats can be stored
//! in files on disk or object storage.
+use std::io::Cursor;
use std::sync::Arc;
-use std::{any::Any, io::Cursor};
use datafusion_datasource::{TableSchema, as_file_source};
@@ -314,10 +314,6 @@ impl FileSource for ArrowSource {
)
}
- fn as_any(&self) -> &dyn Any {
- self
- }
-
fn with_batch_size(&self, _batch_size: usize) -> Arc<dyn FileSource> {
Arc::new(Self { ..self.clone() })
}
diff --git a/datafusion/datasource-avro/src/file_format.rs b/datafusion/datasource-avro/src/file_format.rs
index 35fd90524d..9d9d3279c0 100644
--- a/datafusion/datasource-avro/src/file_format.rs
+++ b/datafusion/datasource-avro/src/file_format.rs
@@ -16,7 +16,6 @@
// under the License.
//! Apache Avro [`FileFormat`] abstractions
-use std::any::Any;
use std::collections::HashMap;
use std::fmt;
use std::sync::Arc;
@@ -65,10 +64,6 @@ impl FileFormatFactory for AvroFormatFactory {
fn default(&self) -> Arc<dyn FileFormat> {
Arc::new(AvroFormat)
}
-
- fn as_any(&self) -> &dyn Any {
- self
- }
}
impl fmt::Debug for AvroFormatFactory {
@@ -90,10 +85,6 @@ pub struct AvroFormat;
#[async_trait]
impl FileFormat for AvroFormat {
- fn as_any(&self) -> &dyn Any {
- self
- }
-
fn get_ext(&self) -> String {
AvroFormatFactory::new().get_ext()
}
diff --git a/datafusion/datasource-avro/src/source.rs b/datafusion/datasource-avro/src/source.rs
index 06026bedf3..b80d4f462e 100644
--- a/datafusion/datasource-avro/src/source.rs
+++ b/datafusion/datasource-avro/src/source.rs
@@ -17,7 +17,6 @@
//! Execution plan for reading line-delimited Avro files
-use std::any::Any;
use std::sync::Arc;
use arrow::datatypes::{Schema, SchemaRef};
@@ -132,10 +131,6 @@ impl FileSource for AvroSource {
Ok(opener)
}
- fn as_any(&self) -> &dyn Any {
- self
- }
-
fn table_schema(&self) -> &TableSchema {
&self.table_schema
}
diff --git a/datafusion/datasource-csv/src/file_format.rs b/datafusion/datasource-csv/src/file_format.rs
index 7a253d81db..9fdd688037 100644
--- a/datafusion/datasource-csv/src/file_format.rs
+++ b/datafusion/datasource-csv/src/file_format.rs
@@ -17,7 +17,6 @@
//! [`CsvFormat`], Comma Separated Value (CSV) [`FileFormat`] abstractions
-use std::any::Any;
use std::collections::{HashMap, HashSet};
use std::fmt::{self, Debug};
use std::sync::Arc;
@@ -122,10 +121,6 @@ impl FileFormatFactory for CsvFormatFactory {
fn default(&self) -> Arc<dyn FileFormat> {
Arc::new(CsvFormat::default())
}
-
- fn as_any(&self) -> &dyn Any {
- self
- }
}
impl GetExt for CsvFormatFactory {
@@ -362,10 +357,6 @@ impl Debug for CsvSerializer {
#[async_trait]
impl FileFormat for CsvFormat {
- fn as_any(&self) -> &dyn Any {
- self
- }
-
fn get_ext(&self) -> String {
CsvFormatFactory::new().get_ext()
}
@@ -450,7 +441,6 @@ impl FileFormat for CsvFormat {
// We need to preserve the table_schema from the original source (which includes partition columns)
let csv_source = conf
.file_source
- .as_any()
.downcast_ref::<CsvSource>()
.expect("file_source should be a CsvSource");
let source = Arc::new(csv_source.clone().with_csv_options(csv_options));
@@ -825,10 +815,6 @@ impl FileSink for CsvSink {
#[async_trait]
impl DataSink for CsvSink {
- fn as_any(&self) -> &dyn Any {
- self
- }
-
fn schema(&self) -> &SchemaRef {
self.config.output_schema()
}
diff --git a/datafusion/datasource-csv/src/source.rs b/datafusion/datasource-csv/src/source.rs
index 77a0dc9cf7..611586cee6 100644
--- a/datafusion/datasource-csv/src/source.rs
+++ b/datafusion/datasource-csv/src/source.rs
@@ -19,7 +19,6 @@
use datafusion_datasource::projection::{ProjectionOpener, SplitProjection};
use datafusion_physical_plan::projection::ProjectionExprs;
-use std::any::Any;
use std::fmt;
use std::io::{Read, Seek, SeekFrom};
use std::sync::Arc;
@@ -263,10 +262,6 @@ impl FileSource for CsvSource {
Ok(opener)
}
- fn as_any(&self) -> &dyn Any {
- self
- }
-
fn table_schema(&self) -> &TableSchema {
&self.table_schema
}
diff --git a/datafusion/datasource-json/src/file_format.rs b/datafusion/datasource-json/src/file_format.rs
index 8fe445705a..1854fddfb8 100644
--- a/datafusion/datasource-json/src/file_format.rs
+++ b/datafusion/datasource-json/src/file_format.rs
@@ -17,7 +17,6 @@
//! [`JsonFormat`]: Line delimited and array JSON [`FileFormat`] abstractions
-use std::any::Any;
use std::collections::HashMap;
use std::fmt;
use std::fmt::Debug;
@@ -112,10 +111,6 @@ impl FileFormatFactory for JsonFormatFactory {
fn default(&self) -> Arc<dyn FileFormat> {
Arc::new(JsonFormat::default())
}
-
- fn as_any(&self) -> &dyn Any {
- self
- }
}
impl GetExt for JsonFormatFactory {
@@ -241,10 +236,6 @@ fn infer_schema_from_json_array<R: Read>(
#[async_trait]
impl FileFormat for JsonFormat {
- fn as_any(&self) -> &dyn Any {
- self
- }
-
fn get_ext(&self) -> String {
JsonFormatFactory::new().get_ext()
}
@@ -488,10 +479,6 @@ impl FileSink for JsonSink {
#[async_trait]
impl DataSink for JsonSink {
- fn as_any(&self) -> &dyn Any {
- self
- }
-
fn schema(&self) -> &SchemaRef {
self.config.output_schema()
}
diff --git a/datafusion/datasource-json/src/source.rs b/datafusion/datasource-json/src/source.rs
index 5cab5ea865..2f2f459956 100644
--- a/datafusion/datasource-json/src/source.rs
+++ b/datafusion/datasource-json/src/source.rs
@@ -17,7 +17,6 @@
//! Execution plan for reading JSON files (line-delimited and array formats)
-use std::any::Any;
use std::io::BufReader;
use std::pin::Pin;
use std::sync::Arc;
@@ -201,10 +200,6 @@ impl FileSource for JsonSource {
Ok(opener)
}
- fn as_any(&self) -> &dyn Any {
- self
- }
-
fn table_schema(&self) -> &datafusion_datasource::TableSchema {
&self.table_schema
}
diff --git a/datafusion/datasource-parquet/src/file_format.rs b/datafusion/datasource-parquet/src/file_format.rs
index c4faedf571..7dda7b1b12 100644
--- a/datafusion/datasource-parquet/src/file_format.rs
+++ b/datafusion/datasource-parquet/src/file_format.rs
@@ -17,7 +17,6 @@
//! [`ParquetFormat`]: Parquet [`FileFormat`] abstractions
-use std::any::Any;
use std::cell::RefCell;
use std::fmt::Debug;
use std::ops::Range;
@@ -152,10 +151,6 @@ impl FileFormatFactory for ParquetFormatFactory {
fn default(&self) -> Arc<dyn FileFormat> {
Arc::new(ParquetFormat::default())
}
-
- fn as_any(&self) -> &dyn Any {
- self
- }
}
impl GetExt for ParquetFormatFactory {
@@ -340,10 +335,6 @@ async fn get_file_decryption_properties(
#[async_trait]
impl FileFormat for ParquetFormat {
- fn as_any(&self) -> &dyn Any {
- self
- }
-
fn get_ext(&self) -> String {
ParquetFormatFactory::new().get_ext()
}
@@ -516,7 +507,6 @@ impl FileFormat for ParquetFormat {
let mut source = conf
.file_source()
- .as_any()
.downcast_ref::<ParquetSource>()
.cloned()
.ok_or_else(|| internal_datafusion_err!("Expected ParquetSource"))?;
@@ -1474,10 +1464,6 @@ impl FileSink for ParquetSink {
#[async_trait]
impl DataSink for ParquetSink {
- fn as_any(&self) -> &dyn Any {
- self
- }
-
fn metrics(&self) -> Option<MetricsSet> {
Some(self.metrics.clone_inner())
}
diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs
index 3037cb1afe..a014c8b272 100644
--- a/datafusion/datasource-parquet/src/source.rs
+++ b/datafusion/datasource-parquet/src/source.rs
@@ -16,7 +16,6 @@
// under the License.
//! ParquetSource implementation for reading parquet files
-use std::any::Any;
use std::fmt::Debug;
use std::fmt::Formatter;
use std::sync::Arc;
@@ -184,7 +183,7 @@ use parquet::encryption::decrypt::FileDecryptionProperties;
/// // Split a single DataSourceExec into multiple DataSourceExecs, one for each file
/// let exec = parquet_exec();
/// let data_source = exec.data_source();
-/// let base_config = data_source.as_any().downcast_ref::<FileScanConfig>().unwrap();
+/// let base_config = data_source.downcast_ref::<FileScanConfig>().unwrap();
/// let existing_file_groups = &base_config.file_groups;
/// let new_execs = existing_file_groups
/// .iter()
@@ -584,10 +583,6 @@ impl FileSource for ParquetSource {
}))
}
- fn as_any(&self) -> &dyn Any {
- self
- }
-
fn table_schema(&self) -> &TableSchema {
&self.table_schema
}
diff --git a/datafusion/datasource/src/file.rs b/datafusion/datasource/src/file.rs
index 9c43eeca2b..9b4ae5827a 100644
--- a/datafusion/datasource/src/file.rs
+++ b/datafusion/datasource/src/file.rs
@@ -63,7 +63,7 @@ pub fn as_file_source<T: FileSource + 'static>(source: T) -> Arc<dyn FileSource>
/// * [`ParquetSource`](https://docs.rs/datafusion/latest/datafusion/datasource/physical_plan/struct.ParquetSource.html)
///
/// [`DataSource`]: crate::source::DataSource
-pub trait FileSource: Send + Sync {
+pub trait FileSource: Any + Send + Sync {
/// Creates a `dyn FileOpener` based on given parameters.
///
/// Note: File sources with a native morsel implementation should return an
@@ -91,8 +91,6 @@ pub trait FileSource: Send + Sync {
let opener = self.create_file_opener(object_store, base_config, partition)?;
Ok(Box::new(FileOpenerMorselizer::new(opener)))
}
- /// Any
- fn as_any(&self) -> &dyn Any;
/// Returns the table schema for the overall table (including partition columns, if any)
///
@@ -363,3 +361,15 @@ pub trait FileSource: Send + Sync {
f: &mut dyn FnMut(&dyn PhysicalExpr) -> Result<TreeNodeRecursion>,
) -> Result<TreeNodeRecursion>;
}
+
+impl dyn FileSource {
+ /// Returns `true` if this source is of type `T`.
+ pub fn is<T: FileSource>(&self) -> bool {
+ (self as &dyn Any).is::<T>()
+ }
+
+ /// Attempts to downcast this source to a concrete type `T`.
+ pub fn downcast_ref<T: FileSource>(&self) -> Option<&T> {
+ (self as &dyn Any).downcast_ref()
+ }
+}
diff --git a/datafusion/datasource/src/file_format.rs b/datafusion/datasource/src/file_format.rs
index 9f8fa622d2..dd30881610 100644
--- a/datafusion/datasource/src/file_format.rs
+++ b/datafusion/datasource/src/file_format.rs
@@ -77,11 +77,7 @@ impl FileMeta {
///
/// [`TableProvider`]: https://docs.rs/datafusion/latest/datafusion/catalog/trait.TableProvider.html
#[async_trait]
-pub trait FileFormat: Send + Sync + fmt::Debug {
- /// Returns the table provider as [`Any`] so that it can be
- /// downcast to a specific implementation.
- fn as_any(&self) -> &dyn Any;
-
+pub trait FileFormat: Any + Send + Sync + fmt::Debug {
/// Returns the extension for this FileFormat, e.g. "file.csv" -> csv
fn get_ext(&self) -> String;
@@ -193,10 +189,20 @@ pub trait FileFormat: Send + Sync + fmt::Debug {
fn file_source(&self, table_schema: crate::TableSchema) -> Arc<dyn FileSource>;
}
+impl dyn FileFormat {
+ pub fn is<T: FileFormat>(&self) -> bool {
+ (self as &dyn Any).is::<T>()
+ }
+
+ pub fn downcast_ref<T: FileFormat>(&self) -> Option<&T> {
+ (self as &dyn Any).downcast_ref()
+ }
+}
+
/// Factory for creating [`FileFormat`] instances based on session and command level options
///
/// Users can provide their own `FileFormatFactory` to support arbitrary file formats
-pub trait FileFormatFactory: Sync + Send + GetExt + fmt::Debug {
+pub trait FileFormatFactory: Any + Sync + Send + GetExt + fmt::Debug {
/// Initialize a [FileFormat] and configure based on session and command level options
fn create(
&self,
@@ -206,10 +212,16 @@ pub trait FileFormatFactory: Sync + Send + GetExt + fmt::Debug {
/// Initialize a [FileFormat] with all options set to default values
fn default(&self) -> Arc<dyn FileFormat>;
+}
- /// Returns the table source as [`Any`] so that it can be
- /// downcast to a specific implementation.
- fn as_any(&self) -> &dyn Any;
+impl dyn FileFormatFactory {
+ pub fn is<T: FileFormatFactory>(&self) -> bool {
+ (self as &dyn Any).is::<T>()
+ }
+
+ pub fn downcast_ref<T: FileFormatFactory>(&self) -> Option<&T> {
+ (self as &dyn Any).downcast_ref()
+ }
}
/// A container of [FileFormatFactory] which also implements [FileType].
diff --git a/datafusion/datasource/src/file_scan_config/mod.rs b/datafusion/datasource/src/file_scan_config/mod.rs
index c1afd35da3..3a3277be08 100644
--- a/datafusion/datasource/src/file_scan_config/mod.rs
+++ b/datafusion/datasource/src/file_scan_config/mod.rs
@@ -55,7 +55,7 @@ use datafusion_physical_plan::{
metrics::ExecutionPlanMetricsSet,
};
use log::{debug, warn};
-use std::{any::Any, fmt::Debug, fmt::Formatter, fmt::Result as FmtResult, sync::Arc};
+use std::{fmt::Debug, fmt::Formatter, fmt::Result as FmtResult, sync::Arc};
/// [`FileScanConfig`] represents scanning data from a group of files
///
@@ -75,7 +75,6 @@ use std::{any::Any, fmt::Debug, fmt::Formatter, fmt::Result as FmtResult, sync::
///
/// # Example
/// ```
-/// # use std::any::Any;
/// # use std::sync::Arc;
/// # use arrow::datatypes::{Field, Fields, DataType, Schema, SchemaRef};
/// # use object_store::ObjectStore;
@@ -106,7 +105,6 @@ use std::{any::Any, fmt::Debug, fmt::Formatter, fmt::Result as FmtResult, sync::
/// # };
/// # impl FileSource for ParquetSource {
/// # fn create_file_opener(&self, _: Arc<dyn ObjectStore>, _: &FileScanConfig, _: usize) -> Result<Arc<dyn FileOpener>> { unimplemented!() }
-/// # fn as_any(&self) -> &dyn Any { self }
/// # fn table_schema(&self) -> &TableSchema { &self.table_schema }
/// # fn with_batch_size(&self, _: usize) -> Arc<dyn FileSource> { unimplemented!() }
/// # fn metrics(&self) -> &ExecutionPlanMetricsSet { unimplemented!() }
@@ -597,10 +595,6 @@ impl DataSource for FileScanConfig {
Ok(Box::pin(cooperative(stream)))
}
- fn as_any(&self) -> &dyn Any {
- self
- }
-
fn fmt_as(&self, t: DisplayFormatType, f: &mut Formatter) -> FmtResult {
match t {
DisplayFormatType::Default | DisplayFormatType::Verbose => {
@@ -1407,10 +1401,6 @@ mod tests {
unimplemented!()
}
- fn as_any(&self) -> &dyn Any {
- self
- }
-
fn table_schema(&self) -> &TableSchema {
&self.table_schema
}
@@ -2417,7 +2407,6 @@ mod tests {
panic!("Expected Inexact result");
};
let pushed_config = inner
- .as_any()
.downcast_ref::<FileScanConfig>()
.expect("Expected FileScanConfig");
let pushed_files = pushed_config.file_groups[0].files();
@@ -2430,7 +2419,6 @@ mod tests {
panic!("Expected Inexact result");
};
let pushed_config = inner
- .as_any()
.downcast_ref::<FileScanConfig>()
.expect("Expected FileScanConfig");
let pushed_files = pushed_config.file_groups[0].files();
@@ -2480,10 +2468,6 @@ mod tests {
unimplemented!()
}
- fn as_any(&self) -> &dyn Any {
- self
- }
-
fn table_schema(&self) -> &TableSchema {
&self.table_schema
}
@@ -2542,7 +2526,6 @@ mod tests {
panic!("Expected Inexact result, got {result:?}");
};
let pushed_config = inner
- .as_any()
.downcast_ref::<FileScanConfig>()
.expect("Expected FileScanConfig");
let files = pushed_config.file_groups[0].files();
@@ -2607,7 +2590,6 @@ mod tests {
panic!("Expected Inexact result");
};
let pushed_config = inner
- .as_any()
.downcast_ref::<FileScanConfig>()
.expect("Expected FileScanConfig");
let files = pushed_config.file_groups[0].files();
@@ -2645,7 +2627,6 @@ mod tests {
panic!("Expected Exact result, got {result:?}");
};
let pushed_config = inner
- .as_any()
.downcast_ref::<FileScanConfig>()
.expect("Expected FileScanConfig");
assert!(!pushed_config.output_ordering.is_empty());
@@ -2680,7 +2661,6 @@ mod tests {
panic!("Expected Inexact (downgraded), got {result:?}");
};
let pushed_config = inner
- .as_any()
.downcast_ref::<FileScanConfig>()
.expect("Expected FileScanConfig");
assert!(pushed_config.output_ordering.is_empty());
@@ -2715,7 +2695,6 @@ mod tests {
panic!("Expected Exact result, got {result:?}");
};
let pushed_config = inner
- .as_any()
.downcast_ref::<FileScanConfig>()
.expect("Expected FileScanConfig");
let files = pushed_config.file_groups[0].files();
@@ -2781,7 +2760,6 @@ mod tests {
panic!("Expected Inexact result");
};
let pushed_config = inner
- .as_any()
.downcast_ref::<FileScanConfig>()
.expect("Expected FileScanConfig");
let files0 = pushed_config.file_groups[0].files();
@@ -2822,7 +2800,6 @@ mod tests {
panic!("Expected Inexact result");
};
let pushed_config = inner
- .as_any()
.downcast_ref::<FileScanConfig>()
.expect("Expected FileScanConfig");
let files0 = pushed_config.file_groups[0].files();
@@ -2857,7 +2834,6 @@ mod tests {
panic!("Expected Inexact result");
};
let pushed_config = inner
- .as_any()
.downcast_ref::<FileScanConfig>()
.expect("Expected FileScanConfig");
let files = pushed_config.file_groups[0].files();
@@ -2908,7 +2884,6 @@ mod tests {
panic!("Expected Exact result, got {result:?}");
};
let pushed_config = inner
- .as_any()
.downcast_ref::<FileScanConfig>()
.expect("Expected FileScanConfig");
@@ -2961,7 +2936,6 @@ mod tests {
panic!("Expected Inexact for reverse scan, got {result:?}");
};
let pushed_config = inner
- .as_any()
.downcast_ref::<FileScanConfig>()
.expect("Expected FileScanConfig");
diff --git a/datafusion/datasource/src/memory.rs b/datafusion/datasource/src/memory.rs
index 7115c22080..9f4f8aa0f3 100644
--- a/datafusion/datasource/src/memory.rs
+++ b/datafusion/datasource/src/memory.rs
@@ -15,7 +15,6 @@
// specific language governing permissions and limitations
// under the License.
-use std::any::Any;
use std::cmp::Ordering;
use std::collections::BinaryHeap;
use std::fmt;
@@ -92,10 +91,6 @@ impl DataSource for MemorySourceConfig {
)))
}
- fn as_any(&self) -> &dyn Any {
- self
- }
-
fn fmt_as(&self, t: DisplayFormatType, f: &mut fmt::Formatter) -> fmt::Result {
match t {
DisplayFormatType::Default | DisplayFormatType::Verbose => {
@@ -772,10 +767,6 @@ impl MemSink {
#[async_trait]
impl DataSink for MemSink {
- fn as_any(&self) -> &dyn Any {
- self
- }
-
fn schema(&self) -> &SchemaRef {
&self.schema
}
@@ -930,10 +921,7 @@ mod tests {
.try_swapping_with_projection(&projection)
.unwrap()
.unwrap();
- let new_source = swapped
- .as_any()
- .downcast_ref::<MemorySourceConfig>()
- .unwrap();
+ let new_source = swapped.downcast_ref::<MemorySourceConfig>().unwrap();
assert_eq!(
new_source.fetch,
@@ -1261,9 +1249,8 @@ mod tests {
// Starting = batch(100_000), batch(10_000), batch(100), batch(1).
// It should have split as p1=batch(100_000), p2=[batch(10_000), batch(100), batch(1)]
let partitioned_datasrc = partitioned_datasrc.unwrap();
- let Some(mem_src_config) = partitioned_datasrc
- .as_any()
- .downcast_ref::<MemorySourceConfig>()
+ let Some(mem_src_config) =
+ partitioned_datasrc.downcast_ref::<MemorySourceConfig>()
else {
unreachable!()
};
@@ -1460,9 +1447,8 @@ mod tests {
// Starting = batch(100_000), batch(1), batch(100), batch(10_000).
// It should have split as p1=batch(100_000), p2=[batch(1), batch(100), batch(10_000)]
let partitioned_datasrc = partitioned_datasrc.unwrap();
- let Some(mem_src_config) = partitioned_datasrc
- .as_any()
- .downcast_ref::<MemorySourceConfig>()
+ let Some(mem_src_config) =
+ partitioned_datasrc.downcast_ref::<MemorySourceConfig>()
else {
unreachable!()
};
diff --git a/datafusion/datasource/src/sink.rs b/datafusion/datasource/src/sink.rs
index d438f2de92..2a1f5c4a2f 100644
--- a/datafusion/datasource/src/sink.rs
+++ b/datafusion/datasource/src/sink.rs
@@ -46,11 +46,7 @@ use futures::StreamExt;
/// The `Display` impl is used to format the sink for explain plan
/// output.
#[async_trait]
-pub trait DataSink: DisplayAs + Debug + Send + Sync {
- /// Returns the data sink as [`Any`] so that it can be
- /// downcast to a specific implementation.
- fn as_any(&self) -> &dyn Any;
-
+pub trait DataSink: Any + DisplayAs + Debug + Send + Sync {
/// Return a snapshot of the [MetricsSet] for this
/// [DataSink].
///
@@ -77,6 +73,18 @@ pub trait DataSink: DisplayAs + Debug + Send + Sync {
) -> Result<u64>;
}
+impl dyn DataSink {
+ /// Returns true if the inner type is `T`.
+ pub fn is<T: DataSink>(&self) -> bool {
+ (self as &dyn Any).is::<T>()
+ }
+
+ /// Returns a reference to the inner value as the type `T` if it is of that type.
+ pub fn downcast_ref<T: DataSink>(&self) -> Option<&T> {
+ (self as &dyn Any).downcast_ref()
+ }
+}
+
/// Execution plan for writing record batches to a [`DataSink`]
///
/// Returns a single row with the number of values written
diff --git a/datafusion/datasource/src/source.rs b/datafusion/datasource/src/source.rs
index 81e15d0a2a..02a15502da 100644
--- a/datafusion/datasource/src/source.rs
+++ b/datafusion/datasource/src/source.rs
@@ -37,6 +37,7 @@ use datafusion_physical_plan::{
};
use itertools::Itertools;
+use crate::file::FileSource;
use crate::file_scan_config::FileScanConfig;
use datafusion_common::config::ConfigOptions;
use datafusion_common::tree_node::TreeNodeRecursion;
@@ -122,13 +123,12 @@ use datafusion_physical_plan::filter_pushdown::{
/// │ │
/// └─────────────────────┘
/// ```
-pub trait DataSource: Send + Sync + Debug {
+pub trait DataSource: Any + Send + Sync + Debug {
fn open(
&self,
partition: usize,
context: Arc<TaskContext>,
) -> Result<SendableRecordBatchStream>;
- fn as_any(&self) -> &dyn Any;
/// Format this source for display in explain plans
fn fmt_as(&self, t: DisplayFormatType, f: &mut Formatter) -> fmt::Result;
@@ -248,6 +248,16 @@ pub trait DataSource: Send + Sync + Debug {
}
}
+impl dyn DataSource {
+ pub fn is<T: DataSource>(&self) -> bool {
+ (self as &dyn Any).is::<T>()
+ }
+
+ pub fn downcast_ref<T: DataSource>(&self) -> Option<&T> {
+ (self as &dyn Any).downcast_ref()
+ }
+}
+
/// [`ExecutionPlan`] that reads one or more files
///
/// `DataSourceExec` implements common functionality such as applying
@@ -358,8 +368,7 @@ impl ExecutionPlan for DataSourceExec {
// Add `output_rows_skew` metric to the metrics set.
// Done here because it's a derived metric from output_rows metric.
- if let Some(file_scan_config) =
- self.data_source.as_any().downcast_ref::<FileScanConfig>()
+ if let Some(file_scan_config) = self.data_source.downcast_ref::<FileScanConfig>()
&& file_scan_config.file_source().file_type() == "parquet"
&& let Some(output_rows_skew) =
BaselineMetrics::output_rows_skew_metric(&metrics)
@@ -525,14 +534,14 @@ impl DataSourceExec {
/// Returns `None` if
/// 1. the datasource is not scanning files (`FileScanConfig`)
/// 2. The [`FileScanConfig::file_source`] is not of type `T`
- pub fn downcast_to_file_source<T: 'static>(&self) -> Option<(&FileScanConfig, &T)> {
+ pub fn downcast_to_file_source<T: FileSource>(
+ &self,
+ ) -> Option<(&FileScanConfig, &T)> {
self.data_source()
- .as_any()
.downcast_ref::<FileScanConfig>()
.and_then(|file_scan_conf| {
file_scan_conf
.file_source()
- .as_any()
.downcast_ref::<T>()
.map(|source| (file_scan_conf, source))
})
diff --git a/datafusion/datasource/src/test_util.rs b/datafusion/datasource/src/test_util.rs
index 3a9e78943b..ffd4b6123e 100644
--- a/datafusion/datasource/src/test_util.rs
+++ b/datafusion/datasource/src/test_util.rs
@@ -76,10 +76,6 @@ impl FileSource for MockSource {
unimplemented!()
}
- fn as_any(&self) -> &dyn std::any::Any {
- self
- }
-
fn filter(&self) -> Option<Arc<dyn PhysicalExpr>> {
self.filter.clone()
}
diff --git a/datafusion/expr/src/logical_plan/builder.rs b/datafusion/expr/src/logical_plan/builder.rs
index 188daa724c..017a123eb0 100644
--- a/datafusion/expr/src/logical_plan/builder.rs
+++ b/datafusion/expr/src/logical_plan/builder.rs
@@ -17,7 +17,6 @@
//! This module provides a builder for creating LogicalPlans
-use std::any::Any;
use std::borrow::Cow;
use std::cmp::Ordering;
use std::collections::{HashMap, HashSet};
@@ -2216,10 +2215,6 @@ impl LogicalTableSource {
}
impl TableSource for LogicalTableSource {
- fn as_any(&self) -> &dyn Any {
- self
- }
-
fn schema(&self) -> SchemaRef {
Arc::clone(&self.table_schema)
}
diff --git a/datafusion/expr/src/table_source.rs b/datafusion/expr/src/table_source.rs
index d3b253c0e1..65dce8f3c8 100644
--- a/datafusion/expr/src/table_source.rs
+++ b/datafusion/expr/src/table_source.rs
@@ -91,9 +91,7 @@ impl std::fmt::Display for TableType {
///
/// [`TableProvider`]: https://docs.rs/datafusion/latest/datafusion/datasource/trait.TableProvider.html
/// [`DefaultTableSource`]: https://docs.rs/datafusion/latest/datafusion/datasource/default_table_source/struct.DefaultTableSource.html
-pub trait TableSource: Sync + Send {
- fn as_any(&self) -> &dyn Any;
-
+pub trait TableSource: Any + Sync + Send {
/// Get a reference to the schema for this table
fn schema(&self) -> SchemaRef;
@@ -130,3 +128,13 @@ pub trait TableSource: Sync + Send {
None
}
}
+
+impl dyn TableSource {
+ pub fn is<T: TableSource>(&self) -> bool {
+ (self as &dyn Any).is::<T>()
+ }
+
+ pub fn downcast_ref<T: TableSource>(&self) -> Option<&T> {
+ (self as &dyn Any).downcast_ref()
+ }
+}
diff --git a/datafusion/optimizer/src/push_down_filter.rs b/datafusion/optimizer/src/push_down_filter.rs
index a1a636cfef..e645509907 100644
--- a/datafusion/optimizer/src/push_down_filter.rs
+++ b/datafusion/optimizer/src/push_down_filter.rs
@@ -1466,7 +1466,6 @@ fn contain(e: &Expr, check_map: &HashMap<String, Expr>) -> bool {
#[cfg(test)]
mod tests {
- use std::any::Any;
use std::cmp::Ordering;
use std::fmt::{Debug, Formatter};
@@ -3140,10 +3139,6 @@ mod tests {
.map(|_| self.filter_support.clone())
.collect())
}
-
- fn as_any(&self) -> &dyn Any {
- self
- }
}
fn table_scan_with_pushdown_provider_builder(
diff --git a/datafusion/optimizer/tests/optimizer_integration.rs b/datafusion/optimizer/tests/optimizer_integration.rs
index ed4b0c0410..ad7427acf9 100644
--- a/datafusion/optimizer/tests/optimizer_integration.rs
+++ b/datafusion/optimizer/tests/optimizer_integration.rs
@@ -15,7 +15,6 @@
// specific language governing permissions and limitations
// under the License.
-use std::any::Any;
use std::cmp::Ordering;
use std::collections::HashMap;
use std::fmt::Formatter;
@@ -759,10 +758,6 @@ struct InexactFilterTableSource {
}
impl TableSource for InexactFilterTableSource {
- fn as_any(&self) -> &dyn Any {
- self
- }
-
fn schema(&self) -> SchemaRef {
self.schema.clone()
}
@@ -932,10 +927,6 @@ struct MyTableSource {
}
impl TableSource for MyTableSource {
- fn as_any(&self) -> &dyn Any {
- self
- }
-
fn schema(&self) -> SchemaRef {
self.schema.clone()
}
diff --git a/datafusion/proto/src/logical_plan/file_formats.rs b/datafusion/proto/src/logical_plan/file_formats.rs
index 8df0b3f1d9..4b111a49e4 100644
--- a/datafusion/proto/src/logical_plan/file_formats.rs
+++ b/datafusion/proto/src/logical_plan/file_formats.rs
@@ -215,12 +215,11 @@ impl LogicalExtensionCodec for CsvLogicalExtensionCodec {
buf: &mut Vec<u8>,
node: Arc<dyn FileFormatFactory>,
) -> datafusion_common::Result<()> {
- let options =
- if let Some(csv_factory) = node.as_any().downcast_ref::<CsvFormatFactory>() {
- csv_factory.options.clone().unwrap_or_default()
- } else {
- return exec_err!("{}", "Unsupported FileFormatFactory type".to_string());
- };
+ let options = if let Some(csv_factory) = node.downcast_ref::<CsvFormatFactory>() {
+ csv_factory.options.clone().unwrap_or_default()
+ } else {
+ return exec_err!("{}", "Unsupported FileFormatFactory type".to_string());
+ };
let proto = CsvOptionsProto::from_factory(&CsvFormatFactory {
options: Some(options),
@@ -326,8 +325,7 @@ impl LogicalExtensionCodec for JsonLogicalExtensionCodec {
buf: &mut Vec<u8>,
node: Arc<dyn FileFormatFactory>,
) -> datafusion_common::Result<()> {
- let options = if let Some(json_factory) =
- node.as_any().downcast_ref::<JsonFormatFactory>()
+ let options = if let Some(json_factory) = node.downcast_ref::<JsonFormatFactory>()
{
json_factory.options.clone().unwrap_or_default()
} else {
@@ -674,7 +672,7 @@ mod parquet {
use datafusion_datasource_parquet::file_format::ParquetFormatFactory;
let options = if let Some(parquet_factory) =
- node.as_any().downcast_ref::<ParquetFormatFactory>()
+ node.downcast_ref::<ParquetFormatFactory>()
{
parquet_factory.options.clone().unwrap_or_default()
} else {
diff --git a/datafusion/proto/src/logical_plan/mod.rs b/datafusion/proto/src/logical_plan/mod.rs
index ded656c9ad..9715ecf8d9 100644
--- a/datafusion/proto/src/logical_plan/mod.rs
+++ b/datafusion/proto/src/logical_plan/mod.rs
@@ -254,26 +254,22 @@ impl LogicalExtensionCodec for DefaultLogicalExtensionCodec {
) -> Result<()> {
let mut encoded_file_format = Vec::new();
- let kind = if node.as_any().downcast_ref::<CsvFormatFactory>().is_some() {
+ let kind = if node.downcast_ref::<CsvFormatFactory>().is_some() {
file_formats::CsvLogicalExtensionCodec
.try_encode_file_format(&mut encoded_file_format, Arc::clone(&node))?;
protobuf::FileFormatKind::Csv
- } else if node.as_any().downcast_ref::<JsonFormatFactory>().is_some() {
+ } else if node.downcast_ref::<JsonFormatFactory>().is_some() {
file_formats::JsonLogicalExtensionCodec
.try_encode_file_format(&mut encoded_file_format, Arc::clone(&node))?;
protobuf::FileFormatKind::Json
- } else if node.as_any().downcast_ref::<ArrowFormatFactory>().is_some() {
+ } else if node.downcast_ref::<ArrowFormatFactory>().is_some() {
file_formats::ArrowLogicalExtensionCodec
.try_encode_file_format(&mut encoded_file_format, Arc::clone(&node))?;
protobuf::FileFormatKind::Arrow
} else {
#[cfg(feature = "parquet")]
{
- if node
- .as_any()
- .downcast_ref::<ParquetFormatFactory>()
- .is_some()
- {
+ if node.downcast_ref::<ParquetFormatFactory>().is_some() {
file_formats::ParquetLogicalExtensionCodec.try_encode_file_format(
&mut encoded_file_format,
Arc::clone(&node),
@@ -1132,12 +1128,12 @@ impl AsLogicalPlan for LogicalPlanNode {
serialize_exprs(filters, extension_codec)?;
if let Some(listing_table) = provider.downcast_ref::<ListingTable>() {
- let any = listing_table.options().format.as_any();
+ let format = listing_table.options().format.as_ref();
let file_format_type = {
let mut maybe_some_type = None;
#[cfg(feature = "parquet")]
- if let Some(parquet) = any.downcast_ref::<ParquetFormat>() {
+ if let Some(parquet) = format.downcast_ref::<ParquetFormat>() {
let options = parquet.options();
maybe_some_type =
Some(FileFormatType::Parquet(protobuf::ParquetFormat {
@@ -1145,7 +1141,7 @@ impl AsLogicalPlan for LogicalPlanNode {
}));
};
- if let Some(csv) = any.downcast_ref::<CsvFormat>() {
+ if let Some(csv) = format.downcast_ref::<CsvFormat>() {
let options = csv.options();
maybe_some_type =
Some(FileFormatType::Csv(protobuf::CsvFormat {
@@ -1153,7 +1149,7 @@ impl AsLogicalPlan for LogicalPlanNode {
}));
}
- if let Some(json) = any.downcast_ref::<OtherNdJsonFormat>() {
+ if let Some(json) = format.downcast_ref::<OtherNdJsonFormat>() {
let options = json.options();
maybe_some_type =
Some(FileFormatType::Json(protobuf::NdJsonFormat {
@@ -1162,12 +1158,12 @@ impl AsLogicalPlan for LogicalPlanNode {
}
#[cfg(feature = "avro")]
- if any.is::<AvroFormat>() {
+ if format.is::<AvroFormat>() {
maybe_some_type =
Some(FileFormatType::Avro(protobuf::AvroFormat {}))
}
- if any.is::<ArrowFormat>() {
+ if format.is::<ArrowFormat>() {
maybe_some_type =
Some(FileFormatType::Arrow(protobuf::ArrowFormat {}))
}
diff --git a/datafusion/proto/src/physical_plan/mod.rs b/datafusion/proto/src/physical_plan/mod.rs
index 4e37e9f952..206f4378d3 100644
--- a/datafusion/proto/src/physical_plan/mod.rs
+++ b/datafusion/proto/src/physical_plan/mod.rs
@@ -2868,9 +2868,9 @@ impl protobuf::PhysicalPlanNode {
proto_converter: &dyn PhysicalProtoConverterExtension,
) -> Result<Option<Self>> {
let data_source = data_source_exec.data_source();
- if let Some(maybe_csv) = data_source.as_any().downcast_ref::<FileScanConfig>() {
+ if let Some(maybe_csv) = data_source.downcast_ref::<FileScanConfig>() {
let source = maybe_csv.file_source();
- if let Some(csv_config) = source.as_any().downcast_ref::<CsvSource>() {
+ if let Some(csv_config) = source.downcast_ref::<CsvSource>() {
return Ok(Some(protobuf::PhysicalPlanNode {
physical_plan_type: Some(PhysicalPlanType::CsvScan(
protobuf::CsvScanExecNode {
@@ -2910,9 +2910,9 @@ impl protobuf::PhysicalPlanNode {
}
}
- if let Some(scan_conf) = data_source.as_any().downcast_ref::<FileScanConfig>() {
+ if let Some(scan_conf) = data_source.downcast_ref::<FileScanConfig>() {
let source = scan_conf.file_source();
- if let Some(_json_source) = source.as_any().downcast_ref::<JsonSource>() {
+ if let Some(_json_source) = source.downcast_ref::<JsonSource>() {
return Ok(Some(protobuf::PhysicalPlanNode {
physical_plan_type: Some(PhysicalPlanType::JsonScan(
protobuf::JsonScanExecNode {
@@ -2927,9 +2927,9 @@ impl protobuf::PhysicalPlanNode {
}
}
- if let Some(scan_conf) = data_source.as_any().downcast_ref::<FileScanConfig>() {
+ if let Some(scan_conf) = data_source.downcast_ref::<FileScanConfig>() {
let source = scan_conf.file_source();
- if let Some(_arrow_source) = source.as_any().downcast_ref::<ArrowSource>() {
+ if let Some(_arrow_source) = source.downcast_ref::<ArrowSource>() {
return Ok(Some(protobuf::PhysicalPlanNode {
physical_plan_type: Some(PhysicalPlanType::ArrowScan(
protobuf::ArrowScanExecNode {
@@ -2968,9 +2968,9 @@ impl protobuf::PhysicalPlanNode {
}
#[cfg(feature = "avro")]
- if let Some(maybe_avro) = data_source.as_any().downcast_ref::<FileScanConfig>() {
+ if let Some(maybe_avro) = data_source.downcast_ref::<FileScanConfig>() {
let source = maybe_avro.file_source();
- if source.as_any().downcast_ref::<AvroSource>().is_some() {
+ if source.downcast_ref::<AvroSource>().is_some() {
return Ok(Some(protobuf::PhysicalPlanNode {
physical_plan_type: Some(PhysicalPlanType::AvroScan(
protobuf::AvroScanExecNode {
@@ -2985,9 +2985,7 @@ impl protobuf::PhysicalPlanNode {
}
}
- if let Some(source_conf) =
- data_source.as_any().downcast_ref::<MemorySourceConfig>()
- {
+ if let Some(source_conf) = data_source.downcast_ref::<MemorySourceConfig>() {
let proto_partitions = source_conf
.partitions()
.iter()
@@ -3380,7 +3378,7 @@ impl protobuf::PhysicalPlanNode {
None => None,
};
- if let Some(sink) = exec.sink().as_any().downcast_ref::<JsonSink>() {
+ if let Some(sink) = exec.sink().downcast_ref::<JsonSink>() {
return Ok(Some(protobuf::PhysicalPlanNode {
physical_plan_type: Some(PhysicalPlanType::JsonSink(Box::new(
protobuf::JsonSinkExecNode {
@@ -3393,7 +3391,7 @@ impl protobuf::PhysicalPlanNode {
}));
}
- if let Some(sink) = exec.sink().as_any().downcast_ref::<CsvSink>() {
+ if let Some(sink) = exec.sink().downcast_ref::<CsvSink>() {
return Ok(Some(protobuf::PhysicalPlanNode {
physical_plan_type: Some(PhysicalPlanType::CsvSink(Box::new(
protobuf::CsvSinkExecNode {
@@ -3407,7 +3405,7 @@ impl protobuf::PhysicalPlanNode {
}
#[cfg(feature = "parquet")]
- if let Some(sink) = exec.sink().as_any().downcast_ref::<ParquetSink>() {
+ if let Some(sink) = exec.sink().downcast_ref::<ParquetSink>() {
return Ok(Some(protobuf::PhysicalPlanNode {
physical_plan_type: Some(PhysicalPlanType::ParquetSink(Box::new(
protobuf::ParquetSinkExecNode {
diff --git a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs
index 07bd02c61b..bb91cb029e 100644
--- a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs
+++ b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs
@@ -586,7 +586,6 @@ async fn roundtrip_logical_plan_copy_to_csv() -> Result<()> {
let format_factory = file_type.as_format_factory();
let csv_factory = format_factory
.as_ref()
- .as_any()
.downcast_ref::<CsvFormatFactory>()
.unwrap();
let csv_config = csv_factory.options.as_ref().unwrap();
@@ -655,7 +654,6 @@ async fn roundtrip_logical_plan_copy_to_json() -> Result<()> {
let format_factory = file_type.as_format_factory();
let json_factory = format_factory
.as_ref()
- .as_any()
.downcast_ref::<JsonFormatFactory>()
.unwrap();
let json_config = json_factory.options.as_ref().unwrap();
@@ -727,7 +725,6 @@ async fn roundtrip_logical_plan_copy_to_parquet() -> Result<()> {
let format_factory = file_type.as_format_factory();
let parquet_factory = format_factory
.as_ref()
- .as_any()
.downcast_ref::<ParquetFormatFactory>()
.unwrap();
let parquet_config = parquet_factory.options.as_ref().unwrap();
@@ -781,7 +778,6 @@ async fn roundtrip_default_codec_csv() -> Result<()> {
let csv = dt
.as_format_factory()
.as_ref()
- .as_any()
.downcast_ref::<CsvFormatFactory>()
.unwrap();
let decoded = csv.options.as_ref().unwrap();
@@ -833,7 +829,6 @@ async fn roundtrip_default_codec_json() -> Result<()> {
let json = dt
.as_format_factory()
.as_ref()
- .as_any()
.downcast_ref::<JsonFormatFactory>()
.unwrap();
let decoded = json.options.as_ref().unwrap();
@@ -887,7 +882,6 @@ async fn roundtrip_default_codec_parquet() -> Result<()> {
let pq = dt
.as_format_factory()
.as_ref()
- .as_any()
.downcast_ref::<ParquetFormatFactory>()
.unwrap();
let decoded = pq.options.as_ref().unwrap();
diff --git a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
index 3d5ddc48b9..3f1c306603 100644
--- a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
+++ b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
@@ -974,14 +974,12 @@ fn roundtrip_parquet_exec_attaches_cached_reader_factory_after_roundtrip() -> Re
})?;
let file_scan = data_source
.data_source()
- .as_any()
.downcast_ref::<FileScanConfig>()
.ok_or_else(|| {
internal_datafusion_err!("Expected FileScanConfig after roundtrip")
})?;
let parquet_source = file_scan
.file_source()
- .as_any()
.downcast_ref::<ParquetSource>()
.ok_or_else(|| {
internal_datafusion_err!("Expected ParquetSource after roundtrip")
@@ -1647,11 +1645,7 @@ fn roundtrip_csv_sink() -> Result<()> {
)?;
let roundtrip_plan = roundtrip_plan.downcast_ref::<DataSinkExec>().unwrap();
- let csv_sink = roundtrip_plan
- .sink()
- .as_any()
- .downcast_ref::<CsvSink>()
- .unwrap();
+ let csv_sink = roundtrip_plan.sink().downcast_ref::<CsvSink>().unwrap();
assert_eq!(
CompressionTypeVariant::ZSTD,
csv_sink.writer_options().compression
diff --git a/datafusion/sql/tests/common/mod.rs b/datafusion/sql/tests/common/mod.rs
index 5caade3002..2e91604d33 100644
--- a/datafusion/sql/tests/common/mod.rs
+++ b/datafusion/sql/tests/common/mod.rs
@@ -329,10 +329,6 @@ impl EmptyTable {
}
impl TableSource for EmptyTable {
- fn as_any(&self) -> &dyn Any {
- self
- }
-
fn schema(&self) -> SchemaRef {
Arc::clone(&self.table_schema)
}
diff --git a/docs/source/library-user-guide/upgrading/54.0.0.md b/docs/source/library-user-guide/upgrading/54.0.0.md
index da1db6de32..d5fe197854 100644
--- a/docs/source/library-user-guide/upgrading/54.0.0.md
+++ b/docs/source/library-user-guide/upgrading/54.0.0.md
@@ -201,7 +201,7 @@ let mut stats = Arc::unwrap_or_clone(plan.partition_statistics(None)?);
stats.column_statistics[0].min_value = ...;
```
-### Remove `as_any` from `PhysicalExpr`, `ScalarUDFImpl`, `AggregateUDFImpl`, `WindowUDFImpl`, `ExecutionPlan`, `TableProvider`, `SchemaProvider`, `CatalogProvider`, and `CatalogProviderList`
+### Remove `as_any` from `PhysicalExpr`, `ScalarUDFImpl`, `AggregateUDFImpl`, `WindowUDFImpl`, `ExecutionPlan`, `TableProvider`, `SchemaProvider`, `CatalogProvider`, `CatalogProviderList`, `TableSource`, `FileSource`, `FileFormat`, `FileFormatFactory`, `DataSource`, and `DataSink`
Now that we have a more recent minimum version of Rust, we can take advantage of
trait upcasting. This reduces the amount of boilerplate code that
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org