Improvements to list_files_cache table function (apache#19703)

## Which issue does this PR close?

- Follow on to apache#19616

## Rationale for this change

I had a few minor comments / suggestions while reviewing
apache#19616 from @jizezhang but they
weren't needed to do the initial merge, so I would like to propose them
in a follow up PR

## What changes are included in this PR?

1. Improve documentation
2. Improve handling of `table_ref` in ListingTableURL
3. use Null rather than `"NULL"` in `list_files_cache` table function

I can break this into separate PRs if that would help

## Are these changes tested?

Yes by CI

## Are there any user-facing changes?

The `list_files_cache` function now might return null

Author: alamb

datafusion-cli/src/functions.rs

@@ -703,10 +703,13 @@ impl TableFunctionImpl for StatisticsCacheFunc {
     }
 }
 
-// Implementation of the `list_files_cache` table function in datafusion-cli.
+/// Implementation of the `list_files_cache` table function in datafusion-cli.
+///
+/// This function returns the cached results of running a LIST command on a
+/// particular object store path for a table. The object metadata is returned as
+/// a List of Structs, with one Struct for each object. DataFusion uses these
+/// cached results to plan queries against external tables.
 ///
-/// This function returns the cached results of running a LIST command on a particular object store path for a table. The object metadata is returned as a List of Structs, with one Struct for each object.
-/// DataFusion uses these cached results to plan queries against external tables.
 /// # Schema
 /// ```sql
 /// > describe select * from list_files_cache();
@@ -788,7 +791,7 @@ impl TableFunctionImpl for ListFilesCacheFunc {
             Field::new("metadata", DataType::Struct(nested_fields.clone()), true);
 
         let schema = Arc::new(Schema::new(vec![
-            Field::new("table", DataType::Utf8, false),
+            Field::new("table", DataType::Utf8, true),
             Field::new("path", DataType::Utf8, false),
             Field::new("metadata_size_bytes", DataType::UInt64, false),
             // expires field in ListFilesEntry has type Instant when set, from which we cannot get "the number of seconds", hence using Duration instead of Timestamp as data type.
@@ -821,7 +824,7 @@ impl TableFunctionImpl for ListFilesCacheFunc {
             let mut current_offset: i32 = 0;
 
             for (path, entry) in list_files_cache.list_entries() {
-                table_arr.push(path.table.map_or("NULL".to_string(), |t| t.to_string()));
+                table_arr.push(path.table.map(|t| t.to_string()));
                 path_arr.push(path.path.to_string());
                 metadata_size_bytes_arr.push(entry.size_bytes as u64);
                 // calculates time left before entry expires

datafusion/core/src/datasource/listing_table_factory.rs

@@ -161,9 +161,7 @@ impl TableProviderFactory for ListingTableFactory {
                         }
                         None => format!("*.{}", cmd.file_type.to_lowercase()),
                     };
-                    table_path = table_path
-                        .with_glob(glob.as_ref())?
-                        .with_table_ref(cmd.name.clone());
+                    table_path = table_path.with_glob(glob.as_ref())?;
                 }
                 let schema = options.infer_schema(session_state, &table_path).await?;
                 let df_schema = Arc::clone(&schema).to_dfschema()?;

datafusion/datasource/src/url.rs

@@ -43,7 +43,7 @@ pub struct ListingTableUrl {
     prefix: Path,
     /// An optional glob expression used to filter files
     glob: Option<Pattern>,
-
+    /// Optional table reference for the table this url belongs to
     table_ref: Option<TableReference>,
 }
 
@@ -341,17 +341,19 @@ impl ListingTableUrl {
     }
 
     /// Returns a copy of current [`ListingTableUrl`] with a specified `glob`
-    pub fn with_glob(self, glob: &str) -> Result<Self> {
-        let glob =
-            Pattern::new(glob).map_err(|e| DataFusionError::External(Box::new(e)))?;
-        Self::try_new(self.url, Some(glob))
+    pub fn with_glob(mut self, glob: &str) -> Result<Self> {
+        self.glob =
+            Some(Pattern::new(glob).map_err(|e| DataFusionError::External(Box::new(e)))?);
+        Ok(self)
     }
 
+    /// Set the table reference for this [`ListingTableUrl`]
     pub fn with_table_ref(mut self, table_ref: TableReference) -> Self {
         self.table_ref = Some(table_ref);
         self
     }
 
+    /// Return the table reference for this [`ListingTableUrl`]
     pub fn get_table_ref(&self) -> &Option<TableReference> {
         &self.table_ref
     }

datafusion/execution/src/cache/cache_manager.rs

@@ -196,6 +196,7 @@ pub trait ListFilesCache: CacheAccessor<TableScopedPath, CachedFileList> {
     /// Retrieves the information about the entries currently cached.
     fn list_entries(&self) -> HashMap<TableScopedPath, ListFilesEntry>;
 
+    /// Drop all entries for the given table reference.
     fn drop_table_entries(&self, table_ref: &Option<TableReference>) -> Result<()>;
 }
 

datafusion/execution/src/cache/list_files_cache.rs

@@ -139,6 +139,11 @@ pub const DEFAULT_LIST_FILES_CACHE_MEMORY_LIMIT: usize = 1024 * 1024; // 1MiB
 /// The default cache TTL for the [`DefaultListFilesCache`]
 pub const DEFAULT_LIST_FILES_CACHE_TTL: Option<Duration> = None; // Infinite
 
+/// Key for [`DefaultListFilesCache`]
+///
+/// Each entry is scoped to its use within a specific table so that the cache
+/// can differentiate between identical paths in different tables, and
+/// table-level cache invalidation.
 #[derive(PartialEq, Eq, Hash, Clone, Debug)]
 pub struct TableScopedPath {
     pub table: Option<TableReference>,