diff --git a/datafusion/datasource-parquet/src/metadata.rs b/datafusion/datasource-parquet/src/metadata.rs index 56abf52144028..25639ca72fc7f 100644 --- a/datafusion/datasource-parquet/src/metadata.rs +++ b/datafusion/datasource-parquet/src/metadata.rs @@ -65,11 +65,26 @@ const PARTIAL_NDV_THRESHOLD: f64 = 0.75; /// [`ParquetFileReaderFactory`]: crate::ParquetFileReaderFactory #[derive(Debug)] pub struct DFParquetMetadata<'a> { + /// Source of the Parquet file's bytes. store: &'a dyn ObjectStore, + /// Location, size and last-modified time of the target Parquet file. object_meta: &'a ObjectMeta, + /// Hint for the number of trailing bytes to prefetch before parsing the + /// footer, mirroring [`ParquetMetaDataReader::with_prefetch_hint`]. metadata_size_hint: Option, + /// Decryption properties used to read files encrypted with Parquet + /// Modular Encryption, mirroring + /// [`ParquetMetaDataReader::with_decryption_properties`]. decryption_properties: Option>, + /// Optional cache of previously fetched [`ParquetMetaData`], keyed by + /// file location. file_metadata_cache: Option>, + /// Policy controlling whether the Parquet page index (column and offset + /// indexes) is fetched, mirroring + /// [`ParquetMetaDataReader::with_page_index_policy`]. + /// + /// `None` means the effective policy is chosen automatically, see + /// [`DFParquetMetadata::effective_page_index_policy`]. page_index_policy: Option, /// timeunit to coerce INT96 timestamps to pub coerce_int96: Option, @@ -78,6 +93,10 @@ pub struct DFParquetMetadata<'a> { } impl<'a> DFParquetMetadata<'a> { + /// Create a new `DFParquetMetadata` for the given file. + /// + /// Use the `with_*` builder methods to customize behavior + /// before calling [`Self::fetch_metadata`] or [`Self::fetch_schema`]. pub fn new(store: &'a dyn ObjectStore, object_meta: &'a ObjectMeta) -> Self { Self { store, @@ -91,13 +110,23 @@ impl<'a> DFParquetMetadata<'a> { } } - /// set metadata size hint + /// Set a hint for the number of trailing bytes to prefetch from the end + /// of the file, equivalent to + /// [`ParquetMetaDataReader::with_prefetch_hint`]. + /// + /// Providing a good estimate of the footer (and, if requested, page index) + /// size can save an extra I/O round trip when fetching metadata from the + /// store. pub fn with_metadata_size_hint(mut self, metadata_size_hint: Option) -> Self { self.metadata_size_hint = metadata_size_hint; self } - /// set decryption properties + /// Set the decryption properties used to read an encrypted Parquet file, + /// equivalent to [`ParquetMetaDataReader::with_decryption_properties`]. + /// + /// Only needed when the target file was written with Parquet Modular + /// Encryption. pub fn with_decryption_properties( mut self, decryption_properties: Option>, @@ -106,7 +135,8 @@ impl<'a> DFParquetMetadata<'a> { self } - /// set file metadata cache + /// Set an optional [`FileMetadataCache`] used to avoid re-fetching + /// [`ParquetMetaData`] for files that have already been read. pub fn with_file_metadata_cache( mut self, file_metadata_cache: Option>, @@ -115,7 +145,12 @@ impl<'a> DFParquetMetadata<'a> { self } - /// Sets the policy for loading parquet page index structures (column and offset indexes). + /// Sets the policy for loading parquet page index structures (column and + /// offset indexes), equivalent to + /// [`ParquetMetaDataReader::with_page_index_policy`]. + /// + /// Passing `None` uses a default automatically, based on whether a metadata + /// cache is configured. pub fn with_page_index_policy( mut self, page_index_policy: Option, @@ -124,19 +159,31 @@ impl<'a> DFParquetMetadata<'a> { self } - /// Set timeunit to coerce INT96 timestamps to + /// Set the [`TimeUnit`] that INT96 timestamp columns should be coerced + /// to when reading the schema. + /// + /// INT96 in Parquet has no defined unit or timezone, so leaving this + /// `None` reads INT96 columns as nanosecond timestamps with no timezone + /// — DataFusion's default behavior. pub fn with_coerce_int96(mut self, time_unit: Option) -> Self { self.coerce_int96 = time_unit; self } /// Set the optional timezone applied to INT96-coerced timestamps. + /// + /// Only used when [`Self::with_coerce_int96`] has also been set, and + /// otherwise has no effect. pub fn with_coerce_int96_tz(mut self, timezone: Option>) -> Self { self.coerce_int96_tz = timezone; self } - /// Fetch parquet metadata from the remote object store + /// Fetch the [`ParquetMetaData`] for this file. + /// + /// Consults the [`FileMetadataCache`] first when one is configured and + /// falls back to reading from the object store via + /// [`ParquetMetaDataPushDecoder`] on a cache miss. pub async fn fetch_metadata(&self) -> Result> { // fetch_metadata // │ @@ -193,6 +240,7 @@ impl<'a> DFParquetMetadata<'a> { Ok(metadata) } + /// Resolve the [`PageIndexPolicy`] to use for a fetch. fn effective_page_index_policy(&self, cache_metadata: bool) -> PageIndexPolicy { self.page_index_policy.unwrap_or_else(|| { if cache_metadata && self.file_metadata_cache.is_some() { @@ -203,10 +251,20 @@ impl<'a> DFParquetMetadata<'a> { }) } + /// Check whether `metadata` already has both the column index and the + /// offset index populated (see [`ParquetMetaData::column_index`] and + /// [`ParquetMetaData::offset_index`]). + /// + /// Used to decide whether page index I/O can be skipped. fn metadata_has_page_index(metadata: &ParquetMetaData) -> bool { metadata.column_index().is_some() && metadata.offset_index().is_some() } + /// Store `metadata` in the configured [`FileMetadataCache`], keyed by + /// the file's location. + /// + /// This is a no-op unless a cache has been configured via + /// [`Self::with_file_metadata_cache`]. fn cache_metadata(&self, metadata: Arc) -> Result<()> { if let Some(file_metadata_cache) = &self.file_metadata_cache { file_metadata_cache.put( @@ -220,6 +278,8 @@ impl<'a> DFParquetMetadata<'a> { Ok(()) } + /// Fetch the full [`ParquetMetaData`] (including footer, and optional + /// page index) from the object store. async fn fetch_metadata_from_store( &self, page_index_policy: PageIndexPolicy, @@ -277,6 +337,8 @@ impl<'a> DFParquetMetadata<'a> { Ok(Arc::new(metadata)) } + /// If `metadata` does not already have a page index, fetch and attach the + /// column and offset indexes. async fn load_page_index( store: &dyn ObjectStore, object_meta: &ObjectMeta, @@ -297,7 +359,8 @@ impl<'a> DFParquetMetadata<'a> { Ok(Arc::new(reader.finish().map_err(DataFusionError::from)?)) } - /// Read and parse the schema of the Parquet file + /// Fetch this file's [`ParquetMetaData`] and convert its embedded Thrift + /// schema into an Arrow [`Schema`]. pub async fn fetch_schema(&self) -> Result { let metadata = self.fetch_metadata().await?; @@ -318,7 +381,8 @@ impl<'a> DFParquetMetadata<'a> { Ok(schema) } - /// Return (path, schema) tuple by fetching the schema from Parquet file + /// Convenience wrapper around [`Self::fetch_schema`] that also returns + /// the file's object store [`Path`]. pub(crate) async fn fetch_schema_with_location(&self) -> Result<(Path, Schema)> { let loc_path = self.object_meta.location.clone(); let schema = self.fetch_schema().await?;