Skip to content

Commit a13e269

Browse files
committed
refactor: make file-statistics cache keys schema-aware
File statistics are computed against a specific `file_schema`, but the file-statistics cache was keyed only by table and path. Reading the same path under a different schema could reuse statistics whose `column_statistics` no longer line up, panicking during statistics projection. #22950 worked around this by bypassing the cache entirely for anonymous explicit-schema reads, at the cost of losing cache reuse for them. Introduce a `SchemaFingerprint` (per-column name, type and nullability, derived from `file_schema`) and a `FileStatisticsCacheKey { table, path, schema }`, and key the file-statistics cache on it. Different schemas now get distinct entries (no stale cross-schema reuse) while a repeated read of the same schema reuses its entry, so the #22950 bypass is removed and anonymous explicit-schema reads cache safely again. - The fingerprint excludes field/schema metadata (cannot affect statistics) and partition columns (their statistics are computed separately). - Table-drop invalidation is unchanged: drop_table_entries matches on CacheKey::table_ref(), which still returns the table, so all schema variants for a table are removed together. - The list-files cache continues to key on TableScopedPath. Closes #23072. Signed-off-by: Jiawei Zhao <[email protected]>
1 parent ff677c4 commit a13e269

4 files changed

Lines changed: 164 additions & 26 deletions

File tree

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

Lines changed: 14 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -268,21 +268,6 @@ impl ListingTable {
268268
self
269269
}
270270

271-
fn statistics_cache(
272-
&self,
273-
has_table_reference: bool,
274-
) -> Option<&Arc<FileStatisticsCache>> {
275-
let shared_cache = self.collected_statistics.as_ref()?;
276-
if has_table_reference || self.schema_source == SchemaSource::Inferred {
277-
Some(shared_cache)
278-
} else {
279-
// Anonymous specified-schema reads can use the same file path with
280-
// different logical schemas. File statistics are schema-dependent,
281-
// so avoid reusing stats computed for a different read schema.
282-
None
283-
}
284-
}
285-
286271
/// Specify the SQL definition for this table, if any
287272
pub fn with_definition(mut self, definition: Option<String>) -> Self {
288273
self.definition = definition;
@@ -990,17 +975,24 @@ impl ListingTable {
990975
store: &Arc<dyn ObjectStore>,
991976
part_file: &PartitionedFile,
992977
) -> datafusion_common::Result<(Arc<Statistics>, Option<LexOrdering>)> {
993-
use datafusion_execution::cache::cache_manager::CachedFileMetadata;
978+
use datafusion_execution::cache::cache_manager::{
979+
CachedFileMetadata, FileStatisticsCacheKey, SchemaFingerprint,
980+
};
994981

995-
let path = TableScopedPath {
982+
let key = FileStatisticsCacheKey {
996983
table: part_file.table_reference.clone(),
997984
path: part_file.object_meta.location.clone(),
985+
// Statistics are computed against `file_schema`, so key on it: reads
986+
// of the same path under a different schema get their own entry
987+
// rather than reusing incompatible column statistics.
988+
schema: SchemaFingerprint::from_schema(&self.file_schema),
998989
};
999990
let meta = &part_file.object_meta;
1000991

1001-
// Check cache first - if we have valid cached statistics and ordering
1002-
if let Some(cache) = self.statistics_cache(path.table.is_some())
1003-
&& let Some(cached) = cache.get(&path)
992+
// Check cache first - the schema is part of the key, so a hit is already
993+
// schema-compatible; `is_valid_for` only confirms the file is unchanged.
994+
if let Some(cache) = &self.collected_statistics
995+
&& let Some(cached) = cache.get(&key)
1004996
&& cached.is_valid_for(meta)
1005997
{
1006998
// Return cached statistics and ordering
@@ -1017,9 +1009,9 @@ impl ListingTable {
10171009
let statistics = Arc::new(file_meta.statistics);
10181010

10191011
// Store in cache
1020-
if let Some(cache) = self.statistics_cache(path.table.is_some()) {
1012+
if let Some(cache) = &self.collected_statistics {
10211013
cache.put(
1022-
&path,
1014+
&key,
10231015
CachedFileMetadata::new(
10241016
meta.clone(),
10251017
Arc::clone(&statistics),

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

Lines changed: 23 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -255,7 +255,29 @@ async fn anonymous_parquet_stats_cache_with_explicit_wider_schema() {
255255
let stats = plan.statistics_with_args(&StatisticsArgs::new()).unwrap();
256256
assert_eq!(stats.column_statistics.len(), 2);
257257
assert_eq!(stats.column_statistics[1].null_count, Precision::Exact(1));
258-
assert_eq!(cache.len(), 1);
258+
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);
263+
264+
// Repeat the wider read: same path + same file_schema -> reuse (no new
265+
// entry) and a cache hit. Under #22950's bypass this read could never reuse.
266+
ctx.read_parquet(
267+
&parquet_path,
268+
ParquetReadOptions::default().schema(&wider_schema),
269+
)
270+
.await
271+
.unwrap()
272+
.create_physical_plan()
273+
.await
274+
.unwrap();
275+
assert_eq!(cache.len(), 2);
276+
let hits: usize = cache.list_entries().values().map(|e| e.hits).sum();
277+
assert_eq!(
278+
hits, 1,
279+
"expected a cache hit on the repeat read, got {hits}"
280+
);
259281
}
260282

261283
#[tokio::test]

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

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

1818
use crate::cache::default_cache::DefaultCache;
19-
pub use crate::cache::{Cache, CacheValue, TableScopedPath};
19+
pub use crate::cache::{
20+
Cache, CacheValue, FileStatisticsCacheKey, SchemaFingerprint, TableScopedPath,
21+
};
2022
use datafusion_common::HashMap;
2123
use datafusion_common::heap_size::{DFHeapSize, DFHeapSizeCtx};
2224
use datafusion_common::{Result, Statistics};
@@ -54,7 +56,7 @@ pub const DEFAULT_METADATA_CACHE_LIMIT: usize = 50 * 1024 * 1024; // 50M
5456
/// 3. If invalid or missing, compute new value and call `put(path, new_value)`
5557
///
5658
/// See [`crate::runtime_env::RuntimeEnv`] for more details
57-
pub type FileStatisticsCache = dyn Cache<TableScopedPath, CachedFileMetadata>;
59+
pub type FileStatisticsCache = dyn Cache<FileStatisticsCacheKey, CachedFileMetadata>;
5860

5961
/// A cache for storing the [`ObjectMeta`]s that result from listing a path.
6062
///
@@ -305,7 +307,7 @@ impl CacheManager {
305307
Some(Arc::clone(fsc))
306308
}
307309
None if config.file_statistics_cache_limit > 0 => Some(Arc::new(
308-
DefaultCache::<TableScopedPath, CachedFileMetadata>::new(
310+
DefaultCache::<FileStatisticsCacheKey, CachedFileMetadata>::new(
309311
config.file_statistics_cache_limit,
310312
)
311313
.with_name("DefaultFileStatisticsCache"),

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

Lines changed: 122 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ pub mod lru_queue;
2020

2121
pub mod default_cache;
2222

23+
use datafusion_common::arrow::datatypes::{DataType, Schema};
2324
use datafusion_common::heap_size::{DFHeapSize, DFHeapSizeCtx};
2425
use datafusion_common::instant::Instant;
2526
use datafusion_common::{HashMap, TableReference};
@@ -165,3 +166,124 @@ impl Display for TableScopedPath {
165166
}
166167
}
167168
}
169+
170+
/// A fingerprint of the `file_schema` used to compute a file's statistics.
171+
///
172+
/// Captures exactly the attributes that determine the layout and meaning of
173+
/// `Statistics::column_statistics`: each column's name, data type and
174+
/// nullability, in order. It deliberately excludes field/schema metadata, which
175+
/// cannot affect statistics — including it would needlessly fragment the cache.
176+
#[derive(PartialEq, Eq, Hash, Clone, Debug)]
177+
pub struct SchemaFingerprint(Vec<(String, DataType, bool)>);
178+
179+
impl SchemaFingerprint {
180+
/// Builds a fingerprint from the `file_schema` used to compute statistics
181+
/// (the schema of the columns physically read, not the full table schema —
182+
/// partition columns and their statistics are handled separately).
183+
pub fn from_schema(file_schema: &Schema) -> Self {
184+
Self(
185+
file_schema
186+
.fields()
187+
.iter()
188+
.map(|f| (f.name().clone(), f.data_type().clone(), f.is_nullable()))
189+
.collect(),
190+
)
191+
}
192+
}
193+
194+
impl DFHeapSize for SchemaFingerprint {
195+
fn heap_size(&self, ctx: &mut DFHeapSizeCtx) -> usize {
196+
// `(String, DataType, bool)` has no `DFHeapSize` impl (only 2-tuples do),
197+
// so account for each column by hand. `bool` carries no heap.
198+
self.0.capacity() * size_of::<(String, DataType, bool)>()
199+
+ self
200+
.0
201+
.iter()
202+
.map(|(name, data_type, _)| {
203+
name.heap_size(ctx) + data_type.heap_size(ctx)
204+
})
205+
.sum::<usize>()
206+
}
207+
}
208+
209+
/// Cache key for the file-statistics cache.
210+
///
211+
/// Like [`TableScopedPath`] it is scoped by table and path, but it additionally
212+
/// carries a [`SchemaFingerprint`]. File statistics are computed against a
213+
/// specific `file_schema`, so the same path read under different schemas must
214+
/// not share an entry; the fingerprint keeps those entries distinct while a
215+
/// repeated read of the same schema still reuses its entry.
216+
#[derive(PartialEq, Eq, Hash, Clone, Debug)]
217+
pub struct FileStatisticsCacheKey {
218+
pub table: Option<TableReference>,
219+
pub path: Path,
220+
pub schema: SchemaFingerprint,
221+
}
222+
223+
impl DFHeapSize for FileStatisticsCacheKey {
224+
fn heap_size(&self, ctx: &mut DFHeapSizeCtx) -> usize {
225+
self.path.as_ref().heap_size(ctx)
226+
+ self.table.heap_size(ctx)
227+
+ self.schema.heap_size(ctx)
228+
}
229+
}
230+
231+
impl CacheKey for FileStatisticsCacheKey {
232+
fn size(&self) -> usize {
233+
DFHeapSize::heap_size(self, &mut DFHeapSizeCtx::default())
234+
}
235+
236+
fn table_ref(&self) -> Option<&TableReference> {
237+
self.table.as_ref()
238+
}
239+
}
240+
241+
#[cfg(test)]
242+
mod schema_fingerprint_tests {
243+
use super::*;
244+
use datafusion_common::arrow::datatypes::Field;
245+
246+
fn fp(fields: Vec<Field>) -> SchemaFingerprint {
247+
SchemaFingerprint::from_schema(&Schema::new(fields))
248+
}
249+
250+
/// `from_schema` must capture nullability and field order — the two
251+
/// attributes most easily dropped by a wrong implementation.
252+
#[test]
253+
fn fingerprint_captures_nullability_and_order() {
254+
assert_ne!(
255+
fp(vec![Field::new("id", DataType::Int64, false)]),
256+
fp(vec![Field::new("id", DataType::Int64, true)]),
257+
"nullability must affect the fingerprint",
258+
);
259+
260+
let ab = fp(vec![
261+
Field::new("a", DataType::Int64, false),
262+
Field::new("b", DataType::Utf8, true),
263+
]);
264+
let ba = fp(vec![
265+
Field::new("b", DataType::Utf8, true),
266+
Field::new("a", DataType::Int64, false),
267+
]);
268+
assert_ne!(ab, ba, "field order must affect the fingerprint");
269+
}
270+
271+
/// Metadata must NOT affect the fingerprint: it cannot change column
272+
/// statistics, so including it would needlessly fragment the cache.
273+
#[test]
274+
fn fingerprint_ignores_metadata() {
275+
let plain = fp(vec![Field::new("id", DataType::Int64, false)]);
276+
277+
let field_md = SchemaFingerprint::from_schema(&Schema::new(vec![
278+
Field::new("id", DataType::Int64, false)
279+
.with_metadata([("note".to_string(), "x".to_string())].into()),
280+
]));
281+
assert_eq!(plain, field_md, "field metadata must be ignored");
282+
283+
let schema_md = SchemaFingerprint::from_schema(
284+
&Schema::new(vec![Field::new("id", DataType::Int64, false)])
285+
.with_metadata([("k".to_string(), "v".to_string())].into()),
286+
);
287+
assert_eq!(plain, schema_md, "schema metadata must be ignored");
288+
}
289+
}

0 commit comments

Comments
 (0)