This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-24036-e948f17dbf7c67f753a41639a143f993dc945fe6 in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit e8015e41a7d38098a61c3776debf052ae03ece96 Author: Andrew Lamb <[email protected]> AuthorDate: Tue Aug 4 16:38:16 2026 -0400 refactor: unify `ParquetFileReader` and `CachedParquetFileReader` (#24036) ## Which issue does this PR close? - Related to https://github.com/apache/datafusion/pull/24030 - Related to https://github.com/apache/arrow-rs/issues/9879 ## Rationale for this change Upstream arrow-rs is deprecating the ParquetObjectReader (see https://github.com/apache/arrow-rs/issues/10308) Not using the deprecated code involves copying some code for each existing usage, and I found that `CachedParquetFileReader` and `ParquetFileReader` are almost the same ## What changes are included in this PR? 1. unify `ParquetFileReader` and `CachedParquetFileReader` 2. make the fields non `pub` ## Are these changes tested? Yes by CI ## Are there any user-facing changes? There are breaking API changes, though I think for the better (things are now encapsulated more) --- datafusion/datasource-parquet/src/bloom_filter.rs | 12 +- datafusion/datasource-parquet/src/reader.rs | 156 ++++++++++----------- docs/source/library-user-guide/upgrading/55.0.0.md | 48 +++++++ 3 files changed, 128 insertions(+), 88 deletions(-) diff --git a/datafusion/datasource-parquet/src/bloom_filter.rs b/datafusion/datasource-parquet/src/bloom_filter.rs index 9c3b73e038..a8f01a5547 100644 --- a/datafusion/datasource-parquet/src/bloom_filter.rs +++ b/datafusion/datasource-parquet/src/bloom_filter.rs @@ -250,7 +250,7 @@ mod tests { use datafusion_physical_expr::planner::logical2physical; use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet; use datafusion_pruning::PruningPredicate; - use object_store::ObjectStoreExt; + use object_store::{ObjectStore, ObjectStoreExt}; use parquet::arrow::ArrowWriter; use parquet::arrow::ParquetRecordBatchStreamBuilder; use parquet::arrow::async_reader::ParquetObjectReader; @@ -644,17 +644,15 @@ mod tests { let metrics = ExecutionPlanMetricsSet::new(); let file_metrics = ParquetFileMetrics::new(0, object_meta.location.as_ref(), &metrics); + let store: Arc<dyn ObjectStore> = Arc::new(in_memory); let inner = - ParquetObjectReader::new(Arc::new(in_memory), object_meta.location.clone()) + ParquetObjectReader::new(Arc::clone(&store), object_meta.location.clone()) .with_file_size(object_meta.size); let partitioned_file = PartitionedFile::new_from_meta(object_meta); - let reader = ParquetFileReader { - inner, - file_metrics: file_metrics.clone(), - partitioned_file, - }; + let reader = + ParquetFileReader::new(file_metrics.clone(), store, inner, partitioned_file); let mut builder = ParquetRecordBatchStreamBuilder::new(reader).await.unwrap(); let access_plan = ParquetAccessPlan::new_all(builder.metadata().num_row_groups()); diff --git a/datafusion/datasource-parquet/src/reader.rs b/datafusion/datasource-parquet/src/reader.rs index 4df636b894..ee2d3a17d5 100644 --- a/datafusion/datasource-parquet/src/reader.rs +++ b/datafusion/datasource-parquet/src/reader.rs @@ -86,62 +86,6 @@ impl DefaultParquetFileReaderFactory { } } -/// Implements [`AsyncFileReader`] for a parquet file in object storage. -/// -/// This implementation uses the [`ParquetObjectReader`] to read data from the -/// object store on demand, as required, tracking the number of bytes read. -/// -/// This implementation does not coalesce I/O operations or cache bytes. Such -/// optimizations can be done either at the object store level or by providing a -/// custom implementation of [`ParquetFileReaderFactory`]. -pub struct ParquetFileReader { - pub file_metrics: ParquetFileMetrics, - pub inner: ParquetObjectReader, - pub partitioned_file: PartitionedFile, -} - -impl AsyncFileReader for ParquetFileReader { - fn get_bytes( - &mut self, - range: Range<u64>, - ) -> BoxFuture<'_, parquet::errors::Result<Bytes>> { - let bytes_scanned = range.end - range.start; - self.file_metrics.bytes_scanned.add(bytes_scanned as usize); - self.inner.get_bytes(range) - } - - fn get_byte_ranges( - &mut self, - ranges: Vec<Range<u64>>, - ) -> BoxFuture<'_, parquet::errors::Result<Vec<Bytes>>> - where - Self: Send, - { - let total: u64 = ranges.iter().map(|r| r.end - r.start).sum(); - self.file_metrics.bytes_scanned.add(total as usize); - self.inner.get_byte_ranges(ranges) - } - - fn get_metadata<'a>( - &'a mut self, - options: Option<&'a ArrowReaderOptions>, - ) -> BoxFuture<'a, parquet::errors::Result<Arc<ParquetMetaData>>> { - self.inner.get_metadata(options) - } -} - -impl Drop for ParquetFileReader { - fn drop(&mut self) { - self.file_metrics - .scan_efficiency_ratio - .add_part(self.file_metrics.bytes_scanned.value()); - // Multiple ParquetFileReaders may run, so we set_total to avoid adding the total multiple times - self.file_metrics - .scan_efficiency_ratio - .set_total(self.partitioned_file.object_meta.size as usize); - } -} - impl ParquetFileReaderFactory for DefaultParquetFileReaderFactory { fn create_reader( &self, @@ -166,18 +110,21 @@ impl ParquetFileReaderFactory for DefaultParquetFileReaderFactory { inner = inner.with_footer_size_hint(hint) }; - Ok(Box::new(ParquetFileReader { - inner, + let reader = ParquetFileReader::new( file_metrics, + Arc::clone(&self.store), + inner, partitioned_file, - })) + ) + .with_metadata_hint(metadata_size_hint); + Ok(Box::new(reader)) } } /// Implementation of [`ParquetFileReaderFactory`] supporting the caching of footer and page /// metadata. Reads and updates the [`FileMetadataCache`] with the [`ParquetMetaData`] data. /// -/// [`CachedParquetFileReader::get_metadata`] forwards the [`parquet::file::metadata::PageIndexPolicy`] from +/// [`ParquetFileReader::get_metadata`] forwards the [`parquet::file::metadata::PageIndexPolicy`] from /// [`ArrowReaderOptions`] to [`DFParquetMetadata::fetch_metadata`], so callers such as the /// parquet opener can skip page-index I/O during the initial metadata load. #[derive(Debug)] @@ -223,50 +170,97 @@ impl ParquetFileReaderFactory for CachedParquetFileReaderFactory { inner = inner.with_footer_size_hint(hint) }; - Ok(Box::new(CachedParquetFileReader::new( + let reader = ParquetFileReader::new( file_metrics, Arc::clone(&self.store), inner, partitioned_file, - Arc::clone(&self.metadata_cache), - metadata_size_hint, - ))) + ) + .with_metadata_hint(metadata_size_hint) + .with_metadata_cache(Some(Arc::clone(&self.metadata_cache))); + + Ok(Box::new(reader)) } } -/// Implements [`AsyncFileReader`] for a Parquet file in object storage. Reads the file metadata -/// from the [`FileMetadataCache`], if available, otherwise reads it directly from the file and then -/// updates the cache. -pub struct CachedParquetFileReader { - pub file_metrics: ParquetFileMetrics, +/// Implements [`AsyncFileReader`] for a parquet file in object storage. +/// +/// This implementation uses the [`ParquetObjectReader`] to read data from the +/// object store on demand, as required, tracking the number of bytes read via +/// [`ParquetFileMetrics`]. +/// +/// When configured via [`Self::with_metadata_cache`], [`Self::get_metadata`] +/// reads footer and page metadata from the cache when available and populates +/// the cache otherwise. Without a cache, metadata is fetched fresh on every call. +/// +/// # Notes +/// +/// This implementation does not coalesce I/O operations or cache bytes. Such +/// optimizations can be done either at the object store level or by providing +/// a custom implementation of [`ParquetFileReaderFactory`]. +pub struct ParquetFileReader { + file_metrics: ParquetFileMetrics, store: Arc<dyn ObjectStore>, - pub inner: ParquetObjectReader, + inner: ParquetObjectReader, partitioned_file: PartitionedFile, - metadata_cache: Arc<FileMetadataCache>, + metadata_cache: Option<Arc<FileMetadataCache>>, metadata_size_hint: Option<usize>, } -impl CachedParquetFileReader { - pub fn new( +impl ParquetFileReader { + /// Create a new `ParquetFileReader`. + /// + /// By default the reader has no [`FileMetadataCache`] and no metadata + /// size hint, so metadata is fetched fresh on every call (as + /// [`DefaultParquetFileReaderFactory`] does). Use + /// [`Self::with_metadata_cache`] to read and populate a cache (as + /// [`CachedParquetFileReaderFactory`] does), and + /// [`Self::with_metadata_hint`] to set the size hint. + pub(crate) fn new( file_metrics: ParquetFileMetrics, store: Arc<dyn ObjectStore>, inner: ParquetObjectReader, partitioned_file: PartitionedFile, - metadata_cache: Arc<FileMetadataCache>, - metadata_size_hint: Option<usize>, ) -> Self { Self { file_metrics, store, inner, partitioned_file, - metadata_cache, - metadata_size_hint, + metadata_cache: None, + metadata_size_hint: None, } } + + /// Returns the metrics tracked while reading this file. + pub fn file_metrics(&self) -> &ParquetFileMetrics { + &self.file_metrics + } + + /// Returns the file this reader is reading. + pub fn partitioned_file(&self) -> &PartitionedFile { + &self.partitioned_file + } + + /// Set the [`FileMetadataCache`] for this reader + pub fn with_metadata_cache( + mut self, + metadata_cache: Option<Arc<FileMetadataCache>>, + ) -> Self { + self.metadata_cache = metadata_cache; + self + } + + /// Set the metadata size hint for this reader. + /// + /// See [`DFParquetMetadata::with_metadata_size_hint`] for more details. + pub fn with_metadata_hint(mut self, metadata_size_hint: Option<usize>) -> Self { + self.metadata_size_hint = metadata_size_hint; + self + } } -impl AsyncFileReader for CachedParquetFileReader { +impl AsyncFileReader for ParquetFileReader { fn get_bytes( &mut self, range: Range<u64>, @@ -293,7 +287,7 @@ impl AsyncFileReader for CachedParquetFileReader { options: Option<&'a ArrowReaderOptions>, ) -> BoxFuture<'a, parquet::errors::Result<Arc<ParquetMetaData>>> { let object_meta = self.partitioned_file.object_meta.clone(); - let metadata_cache = Arc::clone(&self.metadata_cache); + let metadata_cache = self.metadata_cache.clone(); async move { #[cfg(feature = "parquet_encryption")] @@ -308,7 +302,7 @@ impl AsyncFileReader for CachedParquetFileReader { DFParquetMetadata::new(&self.store, &object_meta) .with_decryption_properties(file_decryption_properties) - .with_file_metadata_cache(Some(Arc::clone(&metadata_cache))) + .with_file_metadata_cache(metadata_cache) .with_metadata_size_hint(self.metadata_size_hint) .with_page_index_policy(page_index_policy) .fetch_metadata() @@ -324,7 +318,7 @@ impl AsyncFileReader for CachedParquetFileReader { } } -impl Drop for CachedParquetFileReader { +impl Drop for ParquetFileReader { fn drop(&mut self) { self.file_metrics .scan_efficiency_ratio diff --git a/docs/source/library-user-guide/upgrading/55.0.0.md b/docs/source/library-user-guide/upgrading/55.0.0.md index 9a65651fc3..53cba29f9a 100644 --- a/docs/source/library-user-guide/upgrading/55.0.0.md +++ b/docs/source/library-user-guide/upgrading/55.0.0.md @@ -1030,6 +1030,54 @@ The Minimum Supported Rust Version (MSRV) has been updated to [`1.94.0`]. [`1.94.0`]: https://releases.rs/docs/1.94.0/ +### `CachedParquetFileReader` removed; `ParquetFileReader` fields are now private + +`CachedParquetFileReader` duplicated `ParquetFileReader` and has been removed; +`ParquetFileReader`'s fields are also now private, with +`file_metrics()` and `partitioned_file()` accessors added for the two that +were previously public. + +**Who is affected:** + +- Code that names the `CachedParquetFileReader` type. +- Code that constructs a `ParquetFileReader` directly via a struct literal, or + reads/writes its fields. + +**Migration guide:** + +`ParquetFileReader::new` is no longer public; build a reader through +`ParquetFileReaderFactory::create_reader` (via `DefaultParquetFileReaderFactory` +or `CachedParquetFileReaderFactory`) instead of constructing one directly: + +```rust,ignore +// Before +let inner = ParquetObjectReader::new(Arc::clone(&store), location).with_file_size(size); +let reader = CachedParquetFileReader::new( + file_metrics, + store, + inner, + partitioned_file, + metadata_cache, + metadata_size_hint, +); + +// After +let reader = CachedParquetFileReaderFactory::new(store, metadata_cache) + .create_reader(partition_index, partitioned_file, metadata_size_hint, &metrics)?; +``` + +Replace field access with the new accessor methods: + +```rust,ignore +// Before +let bytes_scanned = reader.file_metrics.bytes_scanned.value(); +let location = &reader.partitioned_file.object_meta.location; + +// After +let bytes_scanned = reader.file_metrics().bytes_scanned.value(); +let location = &reader.partitioned_file().object_meta.location; +``` + ### `array_distance` scalar function now rejects multidimensional arrays `array_distance` only supports one-dimensional arrays. Previously, when given --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
