feat: allow downcasting ParquetFileReaderFactory

datafusion/datasource-parquet/src/reader.rs

@@ -68,7 +68,7 @@ pub trait ParquetFileSchemaProvider: Debug + Send + Sync + 'static {
 /// such as pre-cached metadata, I/O coalescing, etc.
 ///
 /// See [`DefaultParquetFileReaderFactory`] for a simple implementation.
-pub trait ParquetFileReaderFactory: Debug + Send + Sync + 'static {
+pub trait ParquetFileReaderFactory: Any + Debug + Send + Sync {
     /// Provides an `AsyncFileReader` for reading data from a parquet file specified
     ///
     /// # Notes
@@ -92,6 +92,18 @@ pub trait ParquetFileReaderFactory: Debug + Send + Sync + 'static {
     ) -> datafusion_common::Result<Box<dyn AsyncFileReader + Send>>;
 }
 
+impl dyn ParquetFileReaderFactory {
+    /// Returns `true` if this factory is of type `T`.
+    pub fn is<T: ParquetFileReaderFactory>(&self) -> bool {
+        (self as &dyn Any).is::<T>()
+    }
+
+    /// Attempts to downcast this factory to a concrete type `T`.
+    pub fn downcast_ref<T: ParquetFileReaderFactory>(&self) -> Option<&T> {
+        (self as &dyn Any).downcast_ref()
+    }
+}
+
 /// Default implementation of [`ParquetFileReaderFactory`]
 ///
 /// This implementation:

datafusion/datasource-parquet/src/source.rs

@@ -1465,6 +1465,40 @@ mod tests {
         assert!(source.filter().is_some());
     }
 
+    #[test]
+    fn parquet_file_reader_factory_downcasts_to_concrete_type() {
+        use crate::CachedParquetFileReaderFactory;
+        use datafusion_execution::cache::cache_manager::FileMetadataCache;
+        use datafusion_execution::cache::default_cache::DefaultCache;
+        use object_store::memory::InMemory;
+
+        let store: Arc<dyn ObjectStore> = Arc::new(InMemory::new());
+        let source = ParquetSource::new(Arc::new(Schema::empty()))
+            .with_parquet_file_reader_factory(Arc::new(
+                DefaultParquetFileReaderFactory::new(Arc::clone(&store)),
+            ));
+        let factory = source.parquet_file_reader_factory().unwrap();
+        assert!(factory.is::<DefaultParquetFileReaderFactory>());
+        assert!(
+            factory
+                .downcast_ref::<CachedParquetFileReaderFactory>()
+                .is_none()
+        );
+
+        let cache: Arc<FileMetadataCache> = Arc::new(DefaultCache::new(1024));
+        let source = ParquetSource::new(Arc::new(Schema::empty()))
+            .with_parquet_file_reader_factory(Arc::new(
+                CachedParquetFileReaderFactory::new(store, cache),
+            ));
+        let factory = source.parquet_file_reader_factory().unwrap();
+        assert!(factory.is::<CachedParquetFileReaderFactory>());
+        assert!(
+            factory
+                .downcast_ref::<DefaultParquetFileReaderFactory>()
+                .is_none()
+        );
+    }
+
     /// Render a `ParquetSource`'s `fmt_extra` output as a `String` for
     /// inspection in tests.
     fn render_fmt_extra(source: &ParquetSource, t: DisplayFormatType) -> String {

docs/source/library-user-guide/upgrading/56.0.0.md

@@ -1306,3 +1306,18 @@ without declaring one, use `FileScanConfigBuilder::with_preserve_order(true)`.
 Code that reads `FileScanConfig::preserve_order` to decide whether files may be
 reordered should call `FileScanConfig::preserves_file_order()` instead, which
 also accounts for the output ordering.
+
+### `ParquetFileReaderFactory` adds `Any` as a supertrait
+
+To enable downcasting of `dyn ParquetFileReaderFactory` to concrete factory
+types (via `is::<T>()` / `downcast_ref::<T>()`), the
+`ParquetFileReaderFactory` trait now has `Any` as a supertrait:
+
+```diff
+- pub trait ParquetFileReaderFactory: Debug + Send + Sync + 'static
++ pub trait ParquetFileReaderFactory: Any + Debug + Send + Sync
+```
+
+The trait already required `'static`, which `Any` implies, so existing
+implementations need no changes. `ParquetSource::parquet_file_reader_factory`
+returns the trait object to downcast.