Add tablescoping for file stats cache

Author: mkleen

datafusion/catalog-listing/src/helpers.rs

@@ -359,7 +359,7 @@ fn try_into_partitioned_file(
 
     let mut pf: PartitionedFile = object_meta.into();
     pf.partition_values = partition_values;
-
+    pf.table_reference = table_path.get_table_ref().clone();
     Ok(pf)
 }
 

datafusion/catalog-listing/src/table.rs

@@ -804,12 +804,12 @@ impl ListingTable {
     ) -> datafusion_common::Result<(Arc<Statistics>, Option<LexOrdering>)> {
         use datafusion_execution::cache::cache_manager::CachedFileMetadata;
 
-        let path = &part_file.object_meta.location;
+        let path = TableScopedPath { table: part_file.table_reference.clone(), path : part_file.object_meta.location.clone()};
         let meta = &part_file.object_meta;
 
         // Check cache first - if we have valid cached statistics and ordering
         if let Some(cache) = &self.collected_statistics
-            && let Some(cached) = cache.get(path)
+            && let Some(cached) = cache.get(&path)
             && cached.is_valid_for(meta)
         {
             // Return cached statistics and ordering
@@ -828,7 +828,7 @@ impl ListingTable {
         // Store in cache
         if let Some(cache) = &self.collected_statistics {
             cache.put(
-                path,
+                &path,
                 CachedFileMetadata::new(
                     meta.clone(),
                     Arc::clone(&statistics),

datafusion/core/src/execution/context/mod.rs

@@ -1427,8 +1427,10 @@ impl SessionContext {
             schema.deregister_table(&table)?;
             if table_type == TableType::Base
                 && let Some(lfc) = self.runtime_env().cache_manager.get_list_files_cache()
+                && let Some(fsc) = self.runtime_env().cache_manager.get_file_statistic_cache()
             {
-                lfc.drop_table_entries(&Some(table_ref))?;
+                lfc.drop_table_entries(&Some(table_ref.clone()))?;
+                fsc.drop_table_entries(&Some(table_ref.clone()))?;
             }
             return Ok(true);
         }

datafusion/datasource/src/mod.rs

@@ -55,7 +55,7 @@ pub use self::url::ListingTableUrl;
 use crate::file_groups::FileGroup;
 use chrono::TimeZone;
 use datafusion_common::stats::Precision;
-use datafusion_common::{ColumnStatistics, Result, exec_datafusion_err};
+use datafusion_common::{ColumnStatistics, Result, exec_datafusion_err, TableReference};
 use datafusion_common::{ScalarValue, Statistics};
 use datafusion_physical_expr::LexOrdering;
 use futures::{Stream, StreamExt};
@@ -151,6 +151,7 @@ pub struct PartitionedFile {
     pub extensions: Option<Arc<dyn std::any::Any + Send + Sync>>,
     /// The estimated size of the parquet metadata, in bytes
     pub metadata_size_hint: Option<usize>,
+    pub table_reference: Option<TableReference>,
 }
 
 impl PartitionedFile {
@@ -170,6 +171,7 @@ impl PartitionedFile {
             ordering: None,
             extensions: None,
             metadata_size_hint: None,
+            table_reference: None
         }
     }
 
@@ -183,6 +185,7 @@ impl PartitionedFile {
             ordering: None,
             extensions: None,
             metadata_size_hint: None,
+            table_reference: None
         }
     }
 
@@ -202,6 +205,7 @@ impl PartitionedFile {
             ordering: None,
             extensions: None,
             metadata_size_hint: None,
+            table_reference: None
         }
         .with_range(start, end)
     }
@@ -213,6 +217,12 @@ impl PartitionedFile {
         self
     }
 
+    pub fn with_table_reference(mut self, table_reference: Option<TableReference>) -> Self {
+        self.table_reference = table_reference;
+        self
+    }
+
+
     /// Size of the file to be scanned (taking into account the range, if present).
     pub fn effective_size(&self) -> u64 {
         if let Some(range) = &self.range {
@@ -339,6 +349,7 @@ impl From<ObjectMeta> for PartitionedFile {
             ordering: None,
             extensions: None,
             metadata_size_hint: None,
+            table_reference: None,
         }
     }
 }
@@ -536,6 +547,7 @@ pub fn generate_test_files(num_files: usize, overlap_factor: f64) -> Vec<FileGro
             ordering: None,
             extensions: None,
             metadata_size_hint: None,
+            table_reference: None,
         };
         files.push(file);
     }

datafusion/execution/src/cache/cache_manager.rs

@@ -95,7 +95,7 @@ impl CachedFileMetadata {
 /// 3. If invalid or missing, compute new value and call `put(path, new_value)`
 ///
 /// See [`crate::runtime_env::RuntimeEnv`] for more details
-pub trait FileStatisticsCache: CacheAccessor<Path, CachedFileMetadata> {
+pub trait FileStatisticsCache: CacheAccessor<TableScopedPath, CachedFileMetadata> {
     /// Cache memory limit in bytes.
     fn cache_limit(&self) -> usize;
 
@@ -104,6 +104,9 @@ pub trait FileStatisticsCache: CacheAccessor<Path, CachedFileMetadata> {
 
     /// Retrieves the information about the entries currently cached.
     fn list_entries(&self) -> HashMap<Path, FileStatisticsCacheEntry>;
+
+    fn drop_table_entries(&self, table_ref: &Option<TableReference>) -> Result<()>;
+
 }
 
 impl DFHeapSize for CachedFileMetadata {