[Parquet Metadata Cache] Document the ListingTable cache (#17133)

* [Parquet Metadata Cache] Document the ListingTable cache

* Apply suggestions from code review

Co-authored-by: Nuno Faria <[email protected]>
Co-authored-by: Jonathan Chen <[email protected]>

---------

Co-authored-by: Nuno Faria <[email protected]>
Co-authored-by: Jonathan Chen <[email protected]>

Author: 3 people

datafusion/core/src/datasource/listing/table.rs

@@ -819,13 +819,26 @@ impl ListingOptions {
     }
 }
 
-/// Reads data from one or more files as a single table.
+/// Built in [`TableProvider`] that reads data from one or more files as a single table.
 ///
-/// Implements [`TableProvider`], a DataFusion data source. The files are read
-/// using an  [`ObjectStore`] instance, for example from local files or objects
-/// from AWS S3.
+/// The files are read using an  [`ObjectStore`] instance, for example from
+/// local files or objects from AWS S3.
+///
+/// # Features:
+/// * Reading multiple files as a single table
+/// * Hive style partitioning (e.g., directories named `date=2024-06-01`)
+/// * Merges schemas from files with compatible but not identical schemas (see [`ListingTableConfig::file_schema`])
+/// * `limit`, `filter` and `projection` pushdown for formats that support it (e.g.,
+///   Parquet)
+/// * Statistics collection and pruning based on file metadata
+/// * Pre-existing sort order (see [`ListingOptions::file_sort_order`])
+/// * Metadata caching to speed up repeated queries (see [`FileMetadataCache`])
+/// * Statistics caching (see [`FileStatisticsCache`])
+///
+/// [`FileMetadataCache`]: datafusion_execution::cache::cache_manager::FileMetadataCache
+///
+/// # Reading Directories and Hive Style Partitioning
 ///
-/// # Reading Directories
 /// For example, given the `table1` directory (or object store prefix)
 ///
 /// ```text
@@ -861,16 +874,23 @@ impl ListingOptions {
 /// If the query has a predicate like `WHERE date = '2024-06-01'`
 /// only the corresponding directory will be read.
 ///
-/// `ListingTable` also supports limit, filter and projection pushdown for formats that
-/// support it as such as Parquet.
-///
 /// # See Also
 ///
 /// 1. [`ListingTableConfig`]: Configuration options
 /// 1. [`DataSourceExec`]: `ExecutionPlan` used by `ListingTable`
 ///
 /// [`DataSourceExec`]: crate::datasource::source::DataSourceExec
 ///
+/// # Caching Metadata
+///
+/// Some formats, such as Parquet, use the `FileMetadataCache` to cache file
+/// metadata that is needed to execute but expensive to read, such as row
+/// groups and statistics. The cache is scoped to the [`SessionContext`] and can
+/// be configured via the [runtime config options].
+///
+/// [`SessionContext`]: crate::prelude::SessionContext
+/// [runtime config options]: https://datafusion.apache.org/user-guide/configs.html#runtime-configuration-settings
+///
 /// # Example: Read a directory of parquet files using a [`ListingTable`]
 ///
 /// ```no_run
@@ -931,10 +951,16 @@ pub struct ListingTable {
     table_schema: SchemaRef,
     /// Indicates how the schema was derived (inferred or explicitly specified)
     schema_source: SchemaSource,
+    /// Options used to configure the listing table such as the file format
+    /// and partitioning information
     options: ListingOptions,
+    /// The SQL definition for this table, if any
     definition: Option<String>,
+    /// Cache for collected file statistics
     collected_statistics: FileStatisticsCache,
+    /// Constraints applied to this table
     constraints: Constraints,
+    /// Column default expressions for columns that are not physically present in the data files
     column_defaults: HashMap<String, Expr>,
     /// Optional [`SchemaAdapterFactory`] for creating schema adapters
     schema_adapter_factory: Option<Arc<dyn SchemaAdapterFactory>>,

datafusion/core/src/lib.rs

@@ -313,17 +313,17 @@
 //! ```
 //!
 //! A [`TableProvider`] provides information for planning and
-//! an [`ExecutionPlan`] for execution. DataFusion includes [`ListingTable`],
-//! a [`TableProvider`] which reads individual files or directories of files
-//! ("partitioned datasets") of the same file format. Users can add
-//! support for new file formats by implementing the [`TableProvider`]
-//! trait.
+//! an [`ExecutionPlan`] for execution. DataFusion includes two built-in
+//! table providers that support common file formats and require no runtime services,
+//! [`ListingTable`] and [`MemTable`]. You can add support for any other data
+//! source and/or file formats by implementing the [`TableProvider`] trait.
 //!
 //! See also:
 //!
 //! 1. [`ListingTable`]: Reads data from one or more Parquet, JSON, CSV, or AVRO
-//!    files supporting HIVE style partitioning, optional compression, directly
-//!    reading from remote object store and more.
+//!    files in one or more local or remote directories. Supports HIVE style
+//!    partitioning, optional compression, directly reading from remote
+//!    object store, file metadata caching, and more.
 //!
 //! 2. [`MemTable`]: Reads data from in memory [`RecordBatch`]es.
 //!

datafusion/execution/src/cache/cache_manager.rs

@@ -25,19 +25,34 @@ use std::collections::HashMap;
 use std::fmt::{Debug, Formatter};
 use std::sync::Arc;
 
-/// The cache of listing files statistics.
-/// if set [`CacheManagerConfig::with_files_statistics_cache`]
-/// Will avoid infer same file statistics repeatedly during the session lifetime,
-/// this cache will store in [`crate::runtime_env::RuntimeEnv`].
+/// A cache for [`Statistics`].
+///
+/// If enabled via [`CacheManagerConfig::with_files_statistics_cache`] this
+/// cache avoids inferring the same file statistics repeatedly during the
+/// session lifetime.
+///
+/// See [`crate::runtime_env::RuntimeEnv`] for more details
 pub type FileStatisticsCache =
     Arc<dyn CacheAccessor<Path, Arc<Statistics>, Extra = ObjectMeta>>;
 
+/// Cache for storing the [`ObjectMeta`]s that result from listing a path
+///
+/// Listing a path means doing an object store "list" operation or `ls`
+/// command on the local filesystem. This operation can be expensive,
+/// especially when done over remote object stores.
+///
+/// See [`crate::runtime_env::RuntimeEnv`] for more details
 pub type ListFilesCache =
     Arc<dyn CacheAccessor<Path, Arc<Vec<ObjectMeta>>, Extra = ObjectMeta>>;
 
-/// Represents generic file-embedded metadata.
+/// Generic file-embedded metadata used with [`FileMetadataCache`].
+///
+/// For example, Parquet footers and page metadata can be represented
+/// using this trait.
+///
+/// See [`crate::runtime_env::RuntimeEnv`] for more details
 pub trait FileMetadata: Any + Send + Sync {
-    /// Returns the file metadata as [`Any`] so that it can be downcasted to a specific
+    /// Returns the file metadata as [`Any`] so that it can be downcast to a specific
     /// implementation.
     fn as_any(&self) -> &dyn Any;
 
@@ -48,7 +63,20 @@ pub trait FileMetadata: Any + Send + Sync {
     fn extra_info(&self) -> HashMap<String, String>;
 }
 
-/// Cache to store file-embedded metadata.
+/// Cache for file-embedded metadata.
+///
+/// This cache stores per-file metadata in the form of [`FileMetadata`],
+///
+/// For example, the built in [`ListingTable`] uses this cache to avoid parsing
+/// Parquet footers multiple times for the same file.
+///
+/// DataFusion provides a default implementation, [`DefaultFilesMetadataCache`],
+/// and users can also provide their own implementations to implement custom
+/// caching strategies.
+///
+/// See [`crate::runtime_env::RuntimeEnv`] for more details.
+///
+/// [`ListingTable`]: https://docs.rs/datafusion/latest/datafusion/datasource/listing/struct.ListingTable.html
 pub trait FileMetadataCache:
     CacheAccessor<ObjectMeta, Arc<dyn FileMetadata>, Extra = ObjectMeta>
 {
@@ -93,6 +121,13 @@ impl Debug for dyn FileMetadataCache {
     }
 }
 
+/// Manages various caches used in DataFusion.
+///
+/// Following DataFusion design principles, DataFusion provides default cache
+/// implementations, while also allowing users to provide their own custom cache
+/// implementations by implementing the relevant traits.
+///
+/// See [`CacheManagerConfig`] for configuration options.
 #[derive(Debug)]
 pub struct CacheManager {
     file_statistic_cache: Option<FileStatisticsCache>,
@@ -130,7 +165,7 @@ impl CacheManager {
         self.file_statistic_cache.clone()
     }
 
-    /// Get the cache of objectMeta under same path.
+    /// Get the cache for storing the result of listing [`ObjectMeta`]s under the same path.
     pub fn get_list_files_cache(&self) -> Option<ListFilesCache> {
         self.list_files_cache.clone()
     }
@@ -181,6 +216,9 @@ impl Default for CacheManagerConfig {
 }
 
 impl CacheManagerConfig {
+    /// Set the cache for files statistics.
+    ///
+    /// Default is `None` (disabled).
     pub fn with_files_statistics_cache(
         mut self,
         cache: Option<FileStatisticsCache>,
@@ -189,11 +227,17 @@ impl CacheManagerConfig {
         self
     }
 
+    /// Set the cache for listing files.
+    ///     
+    /// Default is `None` (disabled).
     pub fn with_list_files_cache(mut self, cache: Option<ListFilesCache>) -> Self {
         self.list_files_cache = cache;
         self
     }
 
+    /// Sets the cache for file-embedded metadata.
+    ///
+    /// Default is a [`DefaultFilesMetadataCache`].
     pub fn with_file_metadata_cache(
         mut self,
         cache: Option<Arc<dyn FileMetadataCache>>,
@@ -202,6 +246,7 @@ impl CacheManagerConfig {
         self
     }
 
+    /// Sets the limit of the file-embedded metadata cache, in bytes.
     pub fn with_metadata_cache_limit(mut self, limit: usize) -> Self {
         self.metadata_cache_limit = limit;
         self

datafusion/execution/src/cache/cache_unit.rs

@@ -30,8 +30,13 @@ use dashmap::DashMap;
 use object_store::path::Path;
 use object_store::ObjectMeta;
 
-/// Collected statistics for files
+/// Default implementation of [`FileStatisticsCache`]
+///
+/// Stores collected statistics for files
+///
 /// Cache is invalided when file size or last modification has changed
+///
+/// [`FileStatisticsCache`]: crate::cache::cache_manager::FileStatisticsCache
 #[derive(Default)]
 pub struct DefaultFileStatisticsCache {
     statistics: DashMap<Path, (ObjectMeta, Arc<Statistics>)>,
@@ -102,8 +107,13 @@ impl CacheAccessor<Path, Arc<Statistics>> for DefaultFileStatisticsCache {
     }
 }
 
+/// Default implementation of [`ListFilesCache`]
+///
 /// Collected files metadata for listing files.
-/// Cache will not invalided until user call remove or clear.
+///
+/// Cache is not invalided until user calls [`Self::remove`] or [`Self::clear`].
+///
+/// [`ListFilesCache`]: crate::cache::cache_manager::ListFilesCache
 #[derive(Default)]
 pub struct DefaultListFilesCache {
     statistics: DashMap<Path, Arc<Vec<ObjectMeta>>>,
@@ -280,21 +290,36 @@ impl DefaultFilesMetadataCacheState {
     }
 }
 
+/// Default implementation of [`FileMetadataCache`]
+///
 /// Collected file embedded metadata cache.
-/// The metadata for some file is invalided when the file size or last modification time have been
-/// changed.
-/// The `memory_limit` passed in the constructor controls the maximum size of the cache, which uses
-/// a Least Recently Used eviction algorithm.
-/// Users should use the `get` and `put` methods. The `get_with_extra` and `put_with_extra` methods
-/// simply call `get` and `put`, respectively.
+///
+/// The metadata for each file is invalidated when the file size or last
+/// modification time have been changed.
+///
+/// # Internal details
+///
+/// The `memory_limit` controls the maximum size of the cache, which uses a
+/// Least Recently Used eviction algorithm. When adding a new entry, if the total
+/// size of the cached entries exceeds `memory_limit`, the least recently used entries
+/// are evicted until the total size is lower than `memory_limit`.
+///
+/// # `Extra` Handling
+///
+/// Users should use the [`Self::get`] and [`Self::put`] methods. The
+/// [`Self::get_with_extra`] and [`Self::put_with_extra`] methods simply call
+/// `get` and `put`, respectively.
 pub struct DefaultFilesMetadataCache {
     // the state is wrapped in a Mutex to ensure the operations are atomic
     state: Mutex<DefaultFilesMetadataCacheState>,
 }
 
 impl DefaultFilesMetadataCache {
-    /// The `memory_limit` parameter controls the maximum size of the cache, in bytes, using a Least
-    /// Recently Used eviction algorithm.
+    /// Create a new instance of [`DefaultFilesMetadataCache`].
+    ///
+    /// # Arguments
+    /// `memory_limit`:  the maximum size of the cache, in bytes
+    //
     pub fn new(memory_limit: usize) -> Self {
         Self {
             state: Mutex::new(DefaultFilesMetadataCacheState::new(memory_limit)),