Skip to content

Commit 985b8f3

Browse files
committed
refactor: validate file stats schema in cache value
Keep the file-statistics cache keyed by table and path while storing the file-schema fingerprint in the cached metadata value. This preserves schema correctness without changing the public cache key type. Signed-off-by: Jiawei Zhao <[email protected]>
1 parent aeb432c commit 985b8f3

6 files changed

Lines changed: 95 additions & 138 deletions

File tree

‎datafusion/catalog-listing/src/table.rs‎

Lines changed: 12 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -38,8 +38,7 @@ use datafusion_datasource::{
3838
ListingTableUrl, PartitionedFile, TableSchemaBuilder, compute_all_files_statistics,
3939
};
4040
use datafusion_execution::cache::cache_manager::{
41-
CachedFileMetadata, FileStatisticsCache, FileStatisticsCacheKey, SchemaFingerprint,
42-
TableScopedPath,
41+
CachedFileMetadata, FileStatisticsCache, SchemaFingerprint, TableScopedPath,
4342
};
4443
use datafusion_expr::dml::InsertOp;
4544
use datafusion_expr::execution_props::ExecutionProps;
@@ -200,8 +199,9 @@ pub struct ListingTable {
200199
column_defaults: HashMap<String, Expr>,
201200
/// Optional [`PhysicalExprAdapterFactory`] for creating physical expression adapters
202201
expr_adapter_factory: Option<Arc<dyn PhysicalExprAdapterFactory>>,
203-
/// Precomputed fingerprint of `file_schema` for the file-statistics cache
204-
/// key. Constant for the table, so computed once here instead of per file.
202+
/// Precomputed fingerprint of `file_schema` for file-statistics cache
203+
/// validation. Constant for the table, so computed once here instead of per
204+
/// file.
205205
file_schema_fingerprint: Arc<SchemaFingerprint>,
206206
}
207207

@@ -985,22 +985,18 @@ impl ListingTable {
985985
store: &Arc<dyn ObjectStore>,
986986
part_file: &PartitionedFile,
987987
) -> datafusion_common::Result<(Arc<Statistics>, Option<LexOrdering>)> {
988-
let key = FileStatisticsCacheKey {
988+
let path = TableScopedPath {
989989
table: part_file.table_reference.clone(),
990990
path: part_file.object_meta.location.clone(),
991-
// Statistics are computed against `file_schema`, so key on its
992-
// fingerprint: reads of the same path under a different schema get
993-
// their own entry rather than reusing incompatible column statistics.
994-
// The fingerprint is precomputed once per table (see `try_new`).
995-
schema: Arc::clone(&self.file_schema_fingerprint),
996991
};
997992
let meta = &part_file.object_meta;
998993

999-
// Check cache first - the schema is part of the key, so a hit is already
1000-
// schema-compatible; `is_valid_for` only confirms the file is unchanged.
994+
// Check cache first. The key stays `{table, path}` for cheap lookups;
995+
// the cached value carries the schema fingerprint to prevent reusing
996+
// stats computed under a different file schema.
1001997
if let Some(cache) = &self.collected_statistics
1002-
&& let Some(cached) = cache.get(&key)
1003-
&& cached.is_valid_for(meta)
998+
&& let Some(cached) = cache.get(&path)
999+
&& cached.is_valid_for(meta, self.file_schema_fingerprint.as_ref())
10041000
{
10051001
// Return cached statistics and ordering
10061002
return Ok((Arc::clone(&cached.statistics), cached.ordering.clone()));
@@ -1018,9 +1014,10 @@ impl ListingTable {
10181014
// Store in cache
10191015
if let Some(cache) = &self.collected_statistics {
10201016
cache.put(
1021-
&key,
1017+
&path,
10221018
CachedFileMetadata::new(
10231019
meta.clone(),
1020+
Arc::clone(&self.file_schema_fingerprint),
10241021
Arc::clone(&statistics),
10251022
file_meta.ordering.clone(),
10261023
),

‎datafusion/core/tests/parquet/file_statistics.rs‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -256,10 +256,10 @@ async fn anonymous_parquet_stats_cache_with_explicit_wider_schema() {
256256
assert_eq!(stats.column_statistics.len(), 2);
257257
assert_eq!(stats.column_statistics[1].null_count, Precision::Exact(1));
258258

259-
// #23072: the cache now keys on file_schema, so the wider read no longer
260-
// bypasses the cache (as in #22950) — it lands in its own entry and
261-
// coexists with the inferred one. Was `1` under the bypass.
262-
assert_eq!(cache.len(), 2);
259+
// #23072: the cache now validates file_schema, so the wider read no
260+
// longer bypasses the cache (as in #22950), but it overwrites the existing
261+
// `{table, path}` entry instead of adding a schema-specific key.
262+
assert_eq!(cache.len(), 1);
263263

264264
// Repeat the wider read: same path + same file_schema -> reuse (no new
265265
// entry) and a cache hit. Under #22950's bypass this read could never reuse.
@@ -272,7 +272,7 @@ async fn anonymous_parquet_stats_cache_with_explicit_wider_schema() {
272272
.create_physical_plan()
273273
.await
274274
.unwrap();
275-
assert_eq!(cache.len(), 2);
275+
assert_eq!(cache.len(), 1);
276276
let hits: usize = cache.list_entries().values().map(|e| e.hits).sum();
277277
assert_eq!(
278278
hits, 1,

‎datafusion/execution/src/cache/cache_manager.rs‎

Lines changed: 19 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -16,9 +16,7 @@
1616
// under the License.
1717

1818
use crate::cache::default_cache::DefaultCache;
19-
pub use crate::cache::{
20-
Cache, CacheValue, FileStatisticsCacheKey, SchemaFingerprint, TableScopedPath,
21-
};
19+
pub use crate::cache::{Cache, CacheValue, SchemaFingerprint, TableScopedPath};
2220
use datafusion_common::HashMap;
2321
use datafusion_common::heap_size::{DFHeapSize, DFHeapSizeCtx};
2422
use datafusion_common::{Result, Statistics};
@@ -52,11 +50,12 @@ pub const DEFAULT_METADATA_CACHE_LIMIT: usize = 50 * 1024 * 1024; // 50M
5250
///
5351
/// The typical usage pattern is:
5452
/// 1. Call `get(path)` to check for cached value
55-
/// 2. If `Some(cached)`, validate with `cached.is_valid_for(&current_meta)`
53+
/// 2. If `Some(cached)`, validate with
54+
/// `cached.is_valid_for(&current_meta, &current_schema_fingerprint)`
5655
/// 3. If invalid or missing, compute new value and call `put(path, new_value)`
5756
///
5857
/// See [`crate::runtime_env::RuntimeEnv`] for more details
59-
pub type FileStatisticsCache = dyn Cache<FileStatisticsCacheKey, CachedFileMetadata>;
58+
pub type FileStatisticsCache = dyn Cache<TableScopedPath, CachedFileMetadata>;
6059

6160
/// A cache for storing the [`ObjectMeta`]s that result from listing a path.
6261
///
@@ -93,11 +92,13 @@ pub type FileMetadataCache = dyn Cache<Path, CachedFileMetadataEntry>;
9392
/// Cached metadata for a file, including statistics and ordering.
9493
///
9594
/// This struct embeds the [`ObjectMeta`] used for cache validation,
96-
/// along with the cached statistics and ordering information.
95+
/// the `file_schema` fingerprint, cached statistics, and ordering information.
9796
#[derive(Debug, Clone, PartialEq, Eq)]
9897
pub struct CachedFileMetadata {
9998
/// File metadata used for cache validation (size, last_modified).
10099
pub meta: ObjectMeta,
100+
/// Fingerprint of the `file_schema` used to compute `statistics`.
101+
pub schema_fingerprint: Arc<SchemaFingerprint>,
101102
/// Cached statistics for the file, if available.
102103
pub statistics: Arc<Statistics>,
103104
/// Cached ordering for the file.
@@ -108,22 +109,29 @@ impl CachedFileMetadata {
108109
/// Create a new cached file metadata entry.
109110
pub fn new(
110111
meta: ObjectMeta,
112+
schema_fingerprint: Arc<SchemaFingerprint>,
111113
statistics: Arc<Statistics>,
112114
ordering: Option<LexOrdering>,
113115
) -> Self {
114116
Self {
115117
meta,
118+
schema_fingerprint,
116119
statistics,
117120
ordering,
118121
}
119122
}
120123

121124
/// Check if this cached entry is still valid for the given metadata.
122125
///
123-
/// Returns true if the file size and last modified time match.
124-
pub fn is_valid_for(&self, current_meta: &ObjectMeta) -> bool {
126+
/// Returns true if the file size, last modified time, and schema match.
127+
pub fn is_valid_for(
128+
&self,
129+
current_meta: &ObjectMeta,
130+
current_schema_fingerprint: &SchemaFingerprint,
131+
) -> bool {
125132
self.meta.size == current_meta.size
126133
&& self.meta.last_modified == current_meta.last_modified
134+
&& self.schema_fingerprint.as_ref() == current_schema_fingerprint
127135
}
128136
}
129137

@@ -141,6 +149,8 @@ impl DFHeapSize for CachedFileMetadata {
141149
+ self.meta.e_tag.heap_size(ctx)
142150
+ self.meta.location.as_ref().heap_size(ctx)
143151
+ self.statistics.heap_size(ctx)
152+
// Do not deep-count `schema_fingerprint`: each ListingTable shares one
153+
// fingerprint across all cached files.
144154
//TODO add ordering once LexOrdering/PhysicalExpr implements DFHeapSize
145155
}
146156
}
@@ -307,7 +317,7 @@ impl CacheManager {
307317
Some(Arc::clone(fsc))
308318
}
309319
None if config.file_statistics_cache_limit > 0 => Some(Arc::new(
310-
DefaultCache::<FileStatisticsCacheKey, CachedFileMetadata>::new(
320+
DefaultCache::<TableScopedPath, CachedFileMetadata>::new(
311321
config.file_statistics_cache_limit,
312322
)
313323
.with_name("DefaultFileStatisticsCache"),

0 commit comments

Comments
 (0)