refactor: make file-statistics cache keys schema-aware (#23201)

## Which issue does this PR close?

<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes #123` indicates that this PR will close issue #123.
-->

- Closes #23072.

## Rationale for this change

<!--
Why are you proposing this change? If this is already explained clearly
in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand
your changes and offer better suggestions for fixes.
-->

File statistics are computed against a specific `file_schema` (their
`column_statistics` are positional, one per column), but the
file-statistics
cache was keyed only by table and path. Reading the same path under a
different
schema could therefore reuse statistics whose columns no longer line up,
panicking during statistics projection.

#22950 worked around this by **bypassing** the file-statistics cache
entirely for
anonymous explicit-schema reads — correct, but it gave up cache reuse
for them
(every such read recomputes statistics). #23072 asks to make the cache
itself
schema-aware so those reads can reuse the cache safely instead of
skipping it.

## What changes are included in this PR?

<!--
There is no need to duplicate the description in the issue here but it
is sometimes worth providing a summary of the individual changes in this
PR.
-->

- Add `SchemaFingerprint` — the per-column `(name, data_type, nullable)`
of a
`file_schema`, in order — and `FileStatisticsCacheKey { table, path,
schema }`,
  and key the file-statistics cache on it (`FileStatisticsCache` is now
  `dyn Cache<FileStatisticsCacheKey, CachedFileMetadata>`).
- `ListingTable::do_collect_statistics_and_ordering` builds the key with
the
`file_schema` fingerprint and uses the shared cache directly. The #22950
bypass (`statistics_cache` helper / `schema_source`-based skip) is
removed:
different schemas now land in distinct entries (no stale cross-schema
reuse),
  while a repeated read of the same schema reuses its entry.
- The fingerprint deliberately **excludes** field/schema metadata (it
cannot
affect statistics, and including it would needlessly fragment the cache)
and
partition columns (partition statistics are computed separately, outside
this
  cache).
- Table-drop invalidation is unchanged: `drop_table_entries` matches on
`CacheKey::table_ref()`, which still returns the table, so all schema
variants
  for a dropped table are removed together.
- The list-files cache continues to key on `TableScopedPath`.

## Are these changes tested?

<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code

If tests are not included in your PR, please explain why (for example,
are they covered by existing tests)?
-->

Yes.

- Updated the #22950 regression test
(`anonymous_parquet_stats_cache_with_explicit_wider_schema`): the wider
explicit-schema read now lands in its own cache entry (2 entries, was 1
under
the bypass) with correct statistics and no panic, and a repeated read of
that
  schema is served from the cache (a cache hit, no new entry).
- Added unit tests for `SchemaFingerprint`: it distinguishes nullability
and
  field order, and ignores field/schema metadata.
- `cargo test` for the `file_statistics` integration module and the
`datafusion-execution` cache tests (including `drop_table_entries`)
pass, along
  with `cargo fmt --all` and `cargo clippy --all-targets --all-features
  -- -D warnings` for the touched crates.

## Are there any user-facing changes?

<!--
If there are user-facing changes then we may require documentation to be
updated before approving the PR.
-->

<!--
If there are any breaking changes to public APIs, please add the `api
change` label.
-->

No change to query results, physical plans, or the serialized (proto)
wire
format; file statistics are computed exactly as before.

One public API change (please add the `api change` label): the
`FileStatisticsCache` type alias now uses `FileStatisticsCacheKey`
instead of
`TableScopedPath` as its key. Code that constructed keys for this cache
directly
must switch to `FileStatisticsCacheKey`. `SchemaFingerprint` and
`FileStatisticsCacheKey` are newly public; `TableScopedPath` remains
(still used
by the list-files cache). `cargo-semver-checks` will flag the key-type
change,
which is expected.

---------

Signed-off-by: Jiawei Zhao <Phoenix500526@163.com>

Author: Phoenix500526

datafusion/catalog-listing/src/table.rs

@@ -37,7 +37,9 @@ use datafusion_datasource::schema_adapter::SchemaAdapterFactory;
 use datafusion_datasource::{
     ListingTableUrl, PartitionedFile, TableSchemaBuilder, compute_all_files_statistics,
 };
-use datafusion_execution::cache::cache_manager::{FileStatisticsCache, TableScopedPath};
+use datafusion_execution::cache::cache_manager::{
+    CachedFileMetadata, FileStatisticsCache, SchemaFingerprint, TableScopedPath,
+};
 use datafusion_expr::dml::InsertOp;
 use datafusion_expr::execution_props::ExecutionProps;
 use datafusion_expr::{
@@ -197,6 +199,10 @@ pub struct ListingTable {
     column_defaults: HashMap<String, Expr>,
     /// Optional [`PhysicalExprAdapterFactory`] for creating physical expression adapters
     expr_adapter_factory: Option<Arc<dyn PhysicalExprAdapterFactory>>,
+    /// Precomputed fingerprint of `file_schema` for file-statistics cache
+    /// validation. Constant for the table, so computed once here instead of per
+    /// file.
+    file_schema_fingerprint: Arc<SchemaFingerprint>,
 }
 
 impl ListingTable {
@@ -227,6 +233,9 @@ impl ListingTable {
                 .with_metadata(file_schema.metadata().clone()),
         );
 
+        let file_schema_fingerprint =
+            Arc::new(SchemaFingerprint::from_schema(&file_schema));
+
         let table = Self {
             table_paths: config.table_paths,
             file_schema,
@@ -238,6 +247,7 @@ impl ListingTable {
             constraints: Constraints::default(),
             column_defaults: HashMap::new(),
             expr_adapter_factory: config.expr_adapter_factory,
+            file_schema_fingerprint,
         };
 
         Ok(table)
@@ -268,21 +278,6 @@ impl ListingTable {
         self
     }
 
-    fn statistics_cache(
-        &self,
-        has_table_reference: bool,
-    ) -> Option<&Arc<FileStatisticsCache>> {
-        let shared_cache = self.collected_statistics.as_ref()?;
-        if has_table_reference || self.schema_source == SchemaSource::Inferred {
-            Some(shared_cache)
-        } else {
-            // Anonymous specified-schema reads can use the same file path with
-            // different logical schemas. File statistics are schema-dependent,
-            // so avoid reusing stats computed for a different read schema.
-            None
-        }
-    }
-
     /// Specify the SQL definition for this table, if any
     pub fn with_definition(mut self, definition: Option<String>) -> Self {
         self.definition = definition;
@@ -990,18 +985,18 @@ impl ListingTable {
         store: &Arc<dyn ObjectStore>,
         part_file: &PartitionedFile,
     ) -> datafusion_common::Result<(Arc<Statistics>, Option<LexOrdering>)> {
-        use datafusion_execution::cache::cache_manager::CachedFileMetadata;
-
         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.statistics_cache(path.table.is_some())
+        // Check cache first. The key stays `{table, path}` for cheap lookups;
+        // the cached value carries the schema fingerprint to prevent reusing
+        // stats computed under a different file schema.
+        if let Some(cache) = &self.collected_statistics
             && let Some(cached) = cache.get(&path)
-            && cached.is_valid_for(meta)
+            && cached.is_valid_for(meta, &self.file_schema_fingerprint)
         {
             // Return cached statistics and ordering
             return Ok((Arc::clone(&cached.statistics), cached.ordering.clone()));
@@ -1017,11 +1012,12 @@ impl ListingTable {
         let statistics = Arc::new(file_meta.statistics);
 
         // Store in cache
-        if let Some(cache) = self.statistics_cache(path.table.is_some()) {
+        if let Some(cache) = &self.collected_statistics {
             cache.put(
                 &path,
                 CachedFileMetadata::new(
                     meta.clone(),
+                    Arc::clone(&self.file_schema_fingerprint),
                     Arc::clone(&statistics),
                     file_meta.ordering.clone(),
                 ),

datafusion/common/src/heap_size.rs

@@ -348,6 +348,17 @@ where
     }
 }
 
+impl<A, B, C> DFHeapSize for (A, B, C)
+where
+    A: DFHeapSize,
+    B: DFHeapSize,
+    C: DFHeapSize,
+{
+    fn heap_size(&self, ctx: &mut DFHeapSizeCtx) -> usize {
+        self.0.heap_size(ctx) + self.1.heap_size(ctx) + self.2.heap_size(ctx)
+    }
+}
+
 impl DFHeapSize for String {
     fn heap_size(&self, _: &mut DFHeapSizeCtx) -> usize {
         self.capacity()

datafusion/core/tests/parquet/file_statistics.rs

@@ -255,7 +255,29 @@ async fn anonymous_parquet_stats_cache_with_explicit_wider_schema() {
     let stats = plan.statistics_with_args(&StatisticsArgs::new()).unwrap();
     assert_eq!(stats.column_statistics.len(), 2);
     assert_eq!(stats.column_statistics[1].null_count, Precision::Exact(1));
+
+    // #23072: the cache now validates file_schema, so the wider read no
+    // longer bypasses the cache (as in #22950), but it overwrites the existing
+    // `{table, path}` entry instead of adding a schema-specific key.
     assert_eq!(cache.len(), 1);
+
+    // Repeat the wider read: same path + same file_schema -> reuse (no new
+    // entry) and a cache hit. Under #22950's bypass this read could never reuse.
+    ctx.read_parquet(
+        &parquet_path,
+        ParquetReadOptions::default().schema(&wider_schema),
+    )
+    .await
+    .unwrap()
+    .create_physical_plan()
+    .await
+    .unwrap();
+    assert_eq!(cache.len(), 1);
+    let hits: usize = cache.list_entries().values().map(|e| e.hits).sum();
+    assert_eq!(
+        hits, 1,
+        "expected a cache hit on the repeat read, got {hits}"
+    );
 }
 
 #[tokio::test]

datafusion/execution/src/cache/cache_manager.rs

@@ -16,7 +16,7 @@
 // under the License.
 
 use crate::cache::default_cache::DefaultCache;
-pub use crate::cache::{Cache, CacheValue, TableScopedPath};
+pub use crate::cache::{Cache, CacheValue, SchemaFingerprint, TableScopedPath};
 use datafusion_common::HashMap;
 use datafusion_common::heap_size::{DFHeapSize, DFHeapSizeCtx};
 use datafusion_common::{Result, Statistics};
@@ -50,7 +50,8 @@ pub const DEFAULT_METADATA_CACHE_LIMIT: usize = 50 * 1024 * 1024; // 50M
 ///
 /// The typical usage pattern is:
 /// 1. Call `get(path)` to check for cached value
-/// 2. If `Some(cached)`, validate with `cached.is_valid_for(&current_meta)`
+/// 2. If `Some(cached)`, validate with
+///    `cached.is_valid_for(&current_meta, &current_schema_fingerprint)`
 /// 3. If invalid or missing, compute new value and call `put(path, new_value)`
 ///
 /// See [`crate::runtime_env::RuntimeEnv`] for more details
@@ -91,11 +92,13 @@ pub type FileMetadataCache = dyn Cache<Path, CachedFileMetadataEntry>;
 /// Cached metadata for a file, including statistics and ordering.
 ///
 /// This struct embeds the [`ObjectMeta`] used for cache validation,
-/// along with the cached statistics and ordering information.
+/// the `file_schema` fingerprint, cached statistics, and ordering information.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct CachedFileMetadata {
     /// File metadata used for cache validation (size, last_modified).
     pub meta: ObjectMeta,
+    /// Fingerprint of the `file_schema` used to compute `statistics`.
+    pub schema_fingerprint: Arc<SchemaFingerprint>,
     /// Cached statistics for the file, if available.
     pub statistics: Arc<Statistics>,
     /// Cached ordering for the file.
@@ -106,22 +109,31 @@ impl CachedFileMetadata {
     /// Create a new cached file metadata entry.
     pub fn new(
         meta: ObjectMeta,
+        schema_fingerprint: Arc<SchemaFingerprint>,
         statistics: Arc<Statistics>,
         ordering: Option<LexOrdering>,
     ) -> Self {
         Self {
             meta,
+            schema_fingerprint,
             statistics,
             ordering,
         }
     }
 
     /// Check if this cached entry is still valid for the given metadata.
     ///
-    /// Returns true if the file size and last modified time match.
-    pub fn is_valid_for(&self, current_meta: &ObjectMeta) -> bool {
+    /// Returns true if the file size, last modified time, and schema match.
+    pub fn is_valid_for(
+        &self,
+        current_meta: &ObjectMeta,
+        current_schema_fingerprint: &Arc<SchemaFingerprint>,
+    ) -> bool {
         self.meta.size == current_meta.size
             && self.meta.last_modified == current_meta.last_modified
+            && (Arc::ptr_eq(&self.schema_fingerprint, current_schema_fingerprint)
+                || self.schema_fingerprint.as_ref()
+                    == current_schema_fingerprint.as_ref())
     }
 }
 
@@ -139,6 +151,8 @@ impl DFHeapSize for CachedFileMetadata {
             + self.meta.e_tag.heap_size(ctx)
             + self.meta.location.as_ref().heap_size(ctx)
             + self.statistics.heap_size(ctx)
+        // Do not deep-count `schema_fingerprint`: each ListingTable shares one
+        // fingerprint across all cached files.
         //TODO add ordering once LexOrdering/PhysicalExpr implements DFHeapSize
     }
 }