You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/07/07 07:02:34 UTC
(datafusion) branch main updated: refactor: make file-statistics cache keys schema-aware (#23201)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 32e11465cd refactor: make file-statistics cache keys schema-aware (#23201)
32e11465cd is described below
commit 32e11465cd038a5080b9e417443d63d05ba633c6
Author: Phoenix <Ph...@163.com>
AuthorDate: Tue Jul 7 14:47:51 2026 +0800
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 <Ph...@163.com>
---
datafusion/catalog-listing/src/table.rs | 40 ++++----
datafusion/common/src/heap_size.rs | 11 ++
datafusion/core/tests/parquet/file_statistics.rs | 22 ++++
datafusion/execution/src/cache/cache_manager.rs | 24 ++++-
datafusion/execution/src/cache/default_cache.rs | 68 +++++++++----
datafusion/execution/src/cache/mod.rs | 111 ++++++++++++++++++++-
docs/source/library-user-guide/upgrading/55.0.0.md | 19 ++++
7 files changed, 249 insertions(+), 46 deletions(-)
diff --git a/datafusion/catalog-listing/src/table.rs b/datafusion/catalog-listing/src/table.rs
index 36d85b981c..9d8b77cfcc 100644
--- a/datafusion/catalog-listing/src/table.rs
+++ b/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(),
),
diff --git a/datafusion/common/src/heap_size.rs b/datafusion/common/src/heap_size.rs
index 869946d824..405736dbf9 100644
--- a/datafusion/common/src/heap_size.rs
+++ b/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()
diff --git a/datafusion/core/tests/parquet/file_statistics.rs b/datafusion/core/tests/parquet/file_statistics.rs
index b082271d67..e0eed40283 100644
--- a/datafusion/core/tests/parquet/file_statistics.rs
+++ b/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]
diff --git a/datafusion/execution/src/cache/cache_manager.rs b/datafusion/execution/src/cache/cache_manager.rs
index cecce81c63..83dcf70975 100644
--- a/datafusion/execution/src/cache/cache_manager.rs
+++ b/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(¤t_meta)`
+/// 2. If `Some(cached)`, validate with
+/// `cached.is_valid_for(¤t_meta, ¤t_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,11 +109,13 @@ 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,
}
@@ -118,10 +123,17 @@ impl CachedFileMetadata {
/// 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
}
}
diff --git a/datafusion/execution/src/cache/default_cache.rs b/datafusion/execution/src/cache/default_cache.rs
index a1d89619eb..bfe326f3a4 100644
--- a/datafusion/execution/src/cache/default_cache.rs
+++ b/datafusion/execution/src/cache/default_cache.rs
@@ -299,7 +299,6 @@ impl<K: CacheKey, V: CacheValue> Cache<K, V> for DefaultCache<K, V> {
mod tests {
use std::sync::Arc;
- use crate::cache::TableScopedPath;
use crate::cache::cache_manager::{
CachedFileList, DEFAULT_LIST_FILES_CACHE_MEMORY_LIMIT, meta_heap_bytes,
};
@@ -311,6 +310,7 @@ mod tests {
use crate::cache::default_cache::TimeProvider;
use crate::cache::{Cache, CacheEntryInfo};
use crate::cache::{CacheKey, CacheValue};
+ use crate::cache::{SchemaFingerprint, TableScopedPath};
use arrow::array::{Int32Array, ListArray, RecordBatch};
use arrow::buffer::{OffsetBuffer, ScalarBuffer};
use arrow::datatypes::{DataType, Field, Schema, TimeUnit};
@@ -831,6 +831,7 @@ mod tests {
path: meta.location.clone(),
table: None,
};
+ let schema_fingerprint = Arc::new(SchemaFingerprint::from_schema(&schema));
// Cache miss
assert!(cache.get(&path).is_none());
@@ -838,6 +839,7 @@ mod tests {
// Put a value
let cached_value = CachedFileMetadata::new(
meta.clone(),
+ Arc::clone(&schema_fingerprint),
Arc::new(Statistics::new_unknown(&schema)),
None,
);
@@ -848,26 +850,39 @@ mod tests {
assert!(result.is_some());
let cached = result.unwrap();
- assert!(cached.is_valid_for(&meta));
+ assert!(cached.is_valid_for(&meta, &schema_fingerprint));
+
+ let equivalent_schema_fingerprint =
+ Arc::new(SchemaFingerprint::from_schema(&schema));
+ assert!(!Arc::ptr_eq(
+ &schema_fingerprint,
+ &equivalent_schema_fingerprint
+ ));
+ assert!(cached.is_valid_for(&meta, &equivalent_schema_fingerprint));
+
+ let different_schema = Schema::new(vec![Field::new(
+ "different_column",
+ DataType::Timestamp(TimeUnit::Second, None),
+ false,
+ )]);
+ let different_schema_fingerprint =
+ Arc::new(SchemaFingerprint::from_schema(&different_schema));
+ assert!(!cached.is_valid_for(&meta, &different_schema_fingerprint));
// File size changed - validation should fail
let meta2 = create_test_meta("test", 2048);
- let path_2 = TableScopedPath {
- path: meta2.location.clone(),
- table: None,
- };
-
- let cached = cache.get(&path_2).unwrap();
- assert!(!cached.is_valid_for(&meta2));
+ let cached = cache.get(&path).unwrap();
+ assert!(!cached.is_valid_for(&meta2, &schema_fingerprint));
// Update with new value
let cached_value2 = CachedFileMetadata::new(
meta2.clone(),
+ Arc::clone(&schema_fingerprint),
Arc::new(Statistics::new_unknown(&schema)),
None,
);
- cache.put(&path_2, cached_value2);
+ cache.put(&path, cached_value2);
// Test list_entries
let entries = cache.list_entries();
@@ -938,10 +953,12 @@ mod tests {
let cache = DefaultCache::new(DEFAULT_FILE_STATISTICS_MEMORY_LIMIT);
let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);
+ let schema_fingerprint = Arc::new(SchemaFingerprint::from_schema(&schema));
// Cache statistics with no ordering
let cached_value = CachedFileMetadata::new(
meta.clone(),
+ Arc::clone(&schema_fingerprint),
Arc::new(Statistics::new_unknown(&schema)),
None, // No ordering yet
);
@@ -958,7 +975,7 @@ mod tests {
// Update to add ordering
let mut cached = cache.get(&path).unwrap();
- if cached.is_valid_for(&meta) && cached.ordering.is_none() {
+ if cached.is_valid_for(&meta, &schema_fingerprint) && cached.ordering.is_none() {
cached.ordering = Some(ordering());
}
cache.put(&path, cached);
@@ -980,12 +997,14 @@ mod tests {
table: None,
};
let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);
+ let schema_fingerprint = Arc::new(SchemaFingerprint::from_schema(&schema));
let meta_v1 = create_test_meta("test.parquet", 100);
// Cache initial value
let cached_value = CachedFileMetadata::new(
meta_v1.clone(),
+ Arc::clone(&schema_fingerprint),
Arc::new(Statistics::new_unknown(&schema)),
None,
);
@@ -996,11 +1015,12 @@ mod tests {
let cached = cache.get(&path).unwrap();
// Should not be valid for new meta
- assert!(!cached.is_valid_for(&meta_v2));
+ assert!(!cached.is_valid_for(&meta_v2, &schema_fingerprint));
// Compute new value and update
let new_cached = CachedFileMetadata::new(
meta_v2.clone(),
+ Arc::clone(&schema_fingerprint),
Arc::new(Statistics::new_unknown(&schema)),
None,
);
@@ -1019,6 +1039,7 @@ mod tests {
table: None,
};
let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);
+ let schema_fingerprint = Arc::new(SchemaFingerprint::from_schema(&schema));
// Cache with original metadata and ordering
let meta_v1 = ObjectMeta {
@@ -1033,6 +1054,7 @@ mod tests {
let ordering_v1 = ordering();
let cached_v1 = CachedFileMetadata::new(
meta_v1.clone(),
+ Arc::clone(&schema_fingerprint),
Arc::new(Statistics::new_unknown(&schema)),
Some(ordering_v1),
);
@@ -1040,7 +1062,7 @@ mod tests {
// Verify cached ordering is valid
let cached = cache.get(&path).unwrap();
- assert!(cached.is_valid_for(&meta_v1));
+ assert!(cached.is_valid_for(&meta_v1, &schema_fingerprint));
assert!(cached.ordering.is_some());
// File modified (size changed)
@@ -1056,12 +1078,13 @@ mod tests {
// Cache entry exists but should be invalid for new metadata
let cached = cache.get(&path).unwrap();
- assert!(!cached.is_valid_for(&meta_v2));
+ assert!(!cached.is_valid_for(&meta_v2, &schema_fingerprint));
// Cache new version with different ordering
let ordering_v2 = ordering(); // New ordering instance
let cached_v2 = CachedFileMetadata::new(
meta_v2.clone(),
+ Arc::clone(&schema_fingerprint),
Arc::new(Statistics::new_unknown(&schema)),
Some(ordering_v2),
);
@@ -1069,10 +1092,10 @@ mod tests {
// Old metadata should be invalid
let cached = cache.get(&path).unwrap();
- assert!(!cached.is_valid_for(&meta_v1));
+ assert!(!cached.is_valid_for(&meta_v1, &schema_fingerprint));
// New metadata should be valid
- assert!(cached.is_valid_for(&meta_v2));
+ assert!(cached.is_valid_for(&meta_v2, &schema_fingerprint));
assert!(cached.ordering.is_some());
}
@@ -1080,11 +1103,13 @@ mod tests {
fn test_list_entries() {
let cache = DefaultCache::new(DEFAULT_FILE_STATISTICS_MEMORY_LIMIT);
let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);
+ let schema_fingerprint = Arc::new(SchemaFingerprint::from_schema(&schema));
let meta1 = create_test_meta("test1.parquet", 100);
let cached_value_1 = CachedFileMetadata::new(
meta1.clone(),
+ Arc::clone(&schema_fingerprint),
Arc::new(Statistics::new_unknown(&schema)),
None,
);
@@ -1098,6 +1123,7 @@ mod tests {
let meta2 = create_test_meta("test2.parquet", 200);
let cached_value_2 = CachedFileMetadata::new(
meta2.clone(),
+ Arc::clone(&schema_fingerprint),
Arc::new(Statistics::new_unknown(&schema)),
Some(ordering()),
);
@@ -1273,8 +1299,14 @@ mod tests {
};
let mut ctx = DFHeapSizeCtx::default();
let object_meta = create_test_meta(file_name, stats.heap_size(&mut ctx) as u64);
- let value =
- CachedFileMetadata::new(object_meta.clone(), Arc::new(stats.clone()), None);
+ let schema = Schema::new(vec![Field::new("list", DataType::Int32, true)]);
+ let schema_fingerprint = Arc::new(SchemaFingerprint::from_schema(&schema));
+ let value = CachedFileMetadata::new(
+ object_meta.clone(),
+ schema_fingerprint,
+ Arc::new(stats.clone()),
+ None,
+ );
(object_meta, value)
}
diff --git a/datafusion/execution/src/cache/mod.rs b/datafusion/execution/src/cache/mod.rs
index 49c2969587..f47a3f3ca4 100644
--- a/datafusion/execution/src/cache/mod.rs
+++ b/datafusion/execution/src/cache/mod.rs
@@ -20,12 +20,14 @@ pub mod lru_queue;
pub mod default_cache;
+use datafusion_common::arrow::datatypes::{DataType, Schema};
use datafusion_common::heap_size::{DFHeapSize, DFHeapSizeCtx};
use datafusion_common::instant::Instant;
use datafusion_common::{HashMap, TableReference};
use object_store::path::Path;
+use std::collections::hash_map::DefaultHasher;
use std::fmt::{Debug, Display, Formatter};
-use std::hash::Hash;
+use std::hash::{Hash, Hasher};
use std::time::Duration;
/// Base trait for cache implementations with common operations.
@@ -165,3 +167,110 @@ impl Display for TableScopedPath {
}
}
}
+
+/// A fingerprint of the `file_schema` used to compute a file's statistics.
+///
+/// Captures exactly the attributes that determine the layout and meaning of
+/// `Statistics::column_statistics`: each column's name, data type and
+/// nullability, in order. It deliberately excludes field/schema metadata, which
+/// cannot affect statistics — including it would needlessly fragment the cache.
+#[derive(Clone, Debug)]
+pub struct SchemaFingerprint {
+ columns: Vec<(String, DataType, bool)>,
+ /// Precomputed hash of `columns`, so hashing a key on every cache lookup is
+ /// O(1) rather than O(schema width). Computed once in `from_schema` with a
+ /// fixed-seed hasher so it is stable across keys; `PartialEq` still compares
+ /// `columns` exactly, so a hash collision can never make two different
+ /// schemas share a cache entry.
+ hash: u64,
+}
+
+impl SchemaFingerprint {
+ /// Builds a fingerprint from the `file_schema` used to compute statistics
+ /// (the schema of the columns physically read, not the full table schema —
+ /// partition columns and their statistics are handled separately).
+ pub fn from_schema(file_schema: &Schema) -> Self {
+ let columns: Vec<(String, DataType, bool)> = file_schema
+ .fields()
+ .iter()
+ .map(|f| (f.name().clone(), f.data_type().clone(), f.is_nullable()))
+ .collect();
+ let mut hasher = DefaultHasher::new();
+ columns.hash(&mut hasher);
+ Self {
+ columns,
+ hash: hasher.finish(),
+ }
+ }
+}
+
+impl PartialEq for SchemaFingerprint {
+ fn eq(&self, other: &Self) -> bool {
+ // Cheap hash gate first, then an exact comparison so collisions are safe.
+ self.hash == other.hash && self.columns == other.columns
+ }
+}
+
+impl Eq for SchemaFingerprint {}
+
+impl Hash for SchemaFingerprint {
+ fn hash<H: Hasher>(&self, state: &mut H) {
+ state.write_u64(self.hash);
+ }
+}
+
+impl DFHeapSize for SchemaFingerprint {
+ fn heap_size(&self, ctx: &mut DFHeapSizeCtx) -> usize {
+ self.columns.heap_size(ctx)
+ }
+}
+
+#[cfg(test)]
+mod schema_fingerprint_tests {
+ use super::*;
+ use datafusion_common::arrow::datatypes::Field;
+
+ fn fp(fields: Vec<Field>) -> SchemaFingerprint {
+ SchemaFingerprint::from_schema(&Schema::new(fields))
+ }
+
+ /// `from_schema` must capture nullability and field order — the two
+ /// attributes most easily dropped by a wrong implementation.
+ #[test]
+ fn fingerprint_captures_nullability_and_order() {
+ assert_ne!(
+ fp(vec![Field::new("id", DataType::Int64, false)]),
+ fp(vec![Field::new("id", DataType::Int64, true)]),
+ "nullability must affect the fingerprint",
+ );
+
+ let ab = fp(vec![
+ Field::new("a", DataType::Int64, false),
+ Field::new("b", DataType::Utf8, true),
+ ]);
+ let ba = fp(vec![
+ Field::new("b", DataType::Utf8, true),
+ Field::new("a", DataType::Int64, false),
+ ]);
+ assert_ne!(ab, ba, "field order must affect the fingerprint");
+ }
+
+ /// Metadata must NOT affect the fingerprint: it cannot change column
+ /// statistics, so including it would needlessly fragment the cache.
+ #[test]
+ fn fingerprint_ignores_metadata() {
+ let plain = fp(vec![Field::new("id", DataType::Int64, false)]);
+
+ let field_md = SchemaFingerprint::from_schema(&Schema::new(vec![
+ Field::new("id", DataType::Int64, false)
+ .with_metadata([("note".to_string(), "x".to_string())].into()),
+ ]));
+ assert_eq!(plain, field_md, "field metadata must be ignored");
+
+ let schema_md = SchemaFingerprint::from_schema(
+ &Schema::new(vec![Field::new("id", DataType::Int64, false)])
+ .with_metadata([("k".to_string(), "v".to_string())].into()),
+ );
+ assert_eq!(plain, schema_md, "schema metadata must be ignored");
+ }
+}
diff --git a/docs/source/library-user-guide/upgrading/55.0.0.md b/docs/source/library-user-guide/upgrading/55.0.0.md
index 307de722bd..a575dbb19f 100644
--- a/docs/source/library-user-guide/upgrading/55.0.0.md
+++ b/docs/source/library-user-guide/upgrading/55.0.0.md
@@ -460,3 +460,22 @@ type aliases:
Implement the newly introduced types for your custom cache implementation.
See [PR #22613](https://github.com/apache/datafusion/pull/22613) for details.
+
+### `CachedFileMetadata` now validates file schema
+
+The file-statistics cache remains keyed by `TableScopedPath`, but
+`CachedFileMetadata` now stores a `SchemaFingerprint` of the `file_schema` used
+to compute the cached statistics. Cache hits are valid only when both the file
+metadata and schema fingerprint match.
+
+**Who is affected:**
+
+- Users constructing `CachedFileMetadata` values directly.
+
+**Migration guide:**
+
+- Pass `Arc::new(SchemaFingerprint::from_schema(file_schema))` to
+ `CachedFileMetadata::new`.
+- Pass the current schema fingerprint to `CachedFileMetadata::is_valid_for`.
+
+See [PR #23201](https://github.com/apache/datafusion/pull/23201) for details.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org