From 177efad169dd616fcf37ed41520ff1c58b995b41 Mon Sep 17 00:00:00 2001 From: Jake Wallin Date: Tue, 20 Oct 2026 11:27:53 +0000 Subject: [PATCH] fix: Validate file cache identity and isolate object stores Scope listing, statistics and Parquet metadata by object store URL and validate size, modification time, ETag/version and schema independently. Invalidate the affected scope when runtime registrations change, while retaining entries for other stores. Preserve FileFormat inference signatures through optional scoped cache configuration. Share store URLs, reuse metadata keys, provide cache-key migration helpers and append the CLI store column to preserve positions. Cover same-URL replacement, deregistration, stale listings, named and anonymous entries, unchanged metadata across stores and CLI column order. --- datafusion-cli/src/functions.rs | 22 +- datafusion-cli/src/main.rs | 120 ++++ datafusion/catalog-listing/src/options.rs | 9 +- datafusion/catalog-listing/src/table.rs | 59 +- .../src/datasource/file_format/parquet.rs | 84 ++- .../tests/datasource/cache_object_identity.rs | 513 ++++++++++++++++++ datafusion/core/tests/datasource/mod.rs | 1 + .../tests/datasource/object_store_access.rs | 41 +- .../datasource-parquet/src/file_format.rs | 37 +- datafusion/datasource-parquet/src/metadata.rs | 136 ++++- .../datasource-parquet/src/opener/mod.rs | 39 +- datafusion/datasource-parquet/src/reader.rs | 26 +- datafusion/datasource-parquet/src/source.rs | 9 +- datafusion/datasource/src/file_format.rs | 14 + datafusion/datasource/src/url.rs | 88 ++- .../execution/src/cache/cache_manager.rs | 260 ++++++++- .../execution/src/cache/default_cache.rs | 336 ++++++++---- datafusion/execution/src/cache/mod.rs | 113 +++- datafusion/execution/src/runtime_env.rs | 5 +- .../library-user-guide/upgrading/56.0.0.md | 40 ++ docs/source/user-guide/cli/functions.md | 81 +-- 21 files changed, 1759 insertions(+), 274 deletions(-) create mode 100644 datafusion/core/tests/datasource/cache_object_identity.rs diff --git a/datafusion-cli/src/functions.rs b/datafusion-cli/src/functions.rs index d5889bd2a50ba..68076b6a8316c 100644 --- a/datafusion-cli/src/functions.rs +++ b/datafusion-cli/src/functions.rs @@ -712,12 +712,14 @@ impl TableFunctionImpl for MetadataCacheFunc { Field::new("metadata_size_bytes", DataType::UInt64, false), Field::new("hits", DataType::UInt64, false), Field::new("extra", DataType::Utf8, true), + Field::new("object_store_url", DataType::Utf8, false), ])); let cached_entries = self.cache_manager.get_file_metadata_cache().list_entries(); // construct record batch from metadata let mut path_arr = Vec::with_capacity(cached_entries.len()); + let mut object_store_url_arr = Vec::with_capacity(cached_entries.len()); let mut file_modified_arr = Vec::with_capacity(cached_entries.len()); let mut file_size_bytes_arr = Vec::with_capacity(cached_entries.len()); let mut e_tag_arr = Vec::with_capacity(cached_entries.len()); @@ -728,7 +730,8 @@ impl TableFunctionImpl for MetadataCacheFunc { let mut extra = Vec::new(); for (path, entry) in cached_entries { - path_arr.push(path.to_string()); + path_arr.push(path.path.to_string()); + object_store_url_arr.push(path.object_store_url.to_string()); file_modified_arr .push(Some(entry.value.meta.last_modified.timestamp_millis())); file_size_bytes_arr.push(entry.value.meta.size); @@ -761,6 +764,7 @@ impl TableFunctionImpl for MetadataCacheFunc { Arc::new(UInt64Array::from(metadata_size_bytes)), Arc::new(UInt64Array::from(hits_arr)), Arc::new(StringArray::from(extra_arr)), + Arc::new(StringArray::from(object_store_url_arr)), ], )?; @@ -835,10 +839,12 @@ impl TableFunctionImpl for StatisticsCacheFunc { Field::new("table_size_bytes", DataType::Utf8, false), Field::new("statistics_size_bytes", DataType::UInt64, false), Field::new("hits", DataType::UInt64, false), + Field::new("object_store_url", DataType::Utf8, false), ])); // construct record batch from metadata let mut path_arr = vec![]; + let mut object_store_url_arr = vec![]; let mut table_arr = vec![]; let mut file_modified_arr = vec![]; let mut file_size_bytes_arr = vec![]; @@ -853,7 +859,8 @@ impl TableFunctionImpl for StatisticsCacheFunc { if let Some(file_statistics_cache) = self.cache_manager.get_file_statistic_cache() { for (path, entry) in file_statistics_cache.list_entries() { - path_arr.push(path.path.to_string()); + path_arr.push(path.store_path.path.to_string()); + object_store_url_arr.push(path.store_path.object_store_url.to_string()); table_arr .push(path.table.map_or_else(|| "".to_string(), |t| t.to_string())); file_modified_arr @@ -891,6 +898,7 @@ impl TableFunctionImpl for StatisticsCacheFunc { Arc::new(StringArray::from(table_size_bytes_arr)), Arc::new(UInt64Array::from(statistics_size_bytes_arr)), Arc::new(UInt64Array::from(hits_arr)), + Arc::new(StringArray::from(object_store_url_arr)), ], )?; @@ -912,11 +920,13 @@ impl TableFunctionImpl for StatisticsCacheFunc { /// +---------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-------------+ /// | column_name | data_type | is_nullable | /// +---------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-------------+ -/// | table | Utf8 | NO | +/// | table | Utf8 | YES | /// | path | Utf8 | NO | /// | metadata_size_bytes | UInt64 | NO | /// | expires_in | Duration(ms) | YES | /// | metadata_list | List(Struct("file_path": non-null Utf8, "file_modified": non-null Timestamp(ms), "file_size_bytes": non-null UInt64, "e_tag": Utf8, "version": Utf8), field: 'metadata') | YES | +/// | hits | UInt64 | NO | +/// | object_store_url | Utf8 | NO | /// +---------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-------------+ /// ``` #[derive(Debug)] @@ -999,10 +1009,12 @@ impl TableFunctionImpl for ListFilesCacheFunc { true, ), Field::new("hits", DataType::UInt64, false), + Field::new("object_store_url", DataType::Utf8, false), ])); let mut table_arr = vec![]; let mut path_arr = vec![]; + let mut object_store_url_arr = vec![]; let mut metadata_size_bytes_arr = vec![]; let mut expires_arr = vec![]; @@ -1020,7 +1032,8 @@ impl TableFunctionImpl for ListFilesCacheFunc { for (path, entry) in list_files_cache.list_entries() { table_arr.push(path.table.map(|t| t.to_string())); - path_arr.push(path.path.to_string()); + path_arr.push(path.store_path.path.to_string()); + object_store_url_arr.push(path.store_path.object_store_url.to_string()); metadata_size_bytes_arr.push(entry.size_bytes as u64); // calculates time left before entry expires expires_arr.push( @@ -1071,6 +1084,7 @@ impl TableFunctionImpl for ListFilesCacheFunc { None, )), Arc::new(UInt64Array::from(hits_arr)), + Arc::new(StringArray::from(object_store_url_arr)), ], )?; diff --git a/datafusion-cli/src/main.rs b/datafusion-cli/src/main.rs index d24f924bfb5bd..2260c658909a2 100644 --- a/datafusion-cli/src/main.rs +++ b/datafusion-cli/src/main.rs @@ -499,6 +499,8 @@ mod tests { use std::time::Duration; use super::*; + use arrow::array::{ArrayRef, RecordBatch, StringArray}; + use arrow::datatypes::DataType; use datafusion::execution::cache::default_cache::DefaultCache; use datafusion::{ common::test_util::batches_to_string, @@ -685,6 +687,124 @@ mod tests { Ok(()) } + #[tokio::test] + async fn test_cache_functions_object_store_identity() -> Result<(), DataFusionError> { + let ctx = SessionContext::new(); + let cache_manager = &ctx.runtime_env().cache_manager; + ctx.register_udtf( + "metadata_cache", + Arc::new(MetadataCacheFunc::new(Arc::clone(cache_manager))), + ); + ctx.register_udtf( + "statistics_cache", + Arc::new(StatisticsCacheFunc::new(Arc::clone(cache_manager))), + ); + ctx.register_udtf( + "list_files_cache", + Arc::new(ListFilesCacheFunc::new(Arc::clone(cache_manager))), + ); + + for store in ["first", "second"] { + ctx.register_object_store( + &Url::parse(&format!("mem://{store}")).unwrap(), + Arc::new(InMemory::new()), + ); + ctx.sql(&format!( + "COPY (SELECT 1 AS value) TO 'mem://{store}/data/file.parquet' STORED AS PARQUET" + )) + .await? + .collect() + .await?; + ctx.read_parquet( + format!("mem://{store}/data/"), + ParquetReadOptions::default(), + ) + .await? + .collect() + .await?; + } + + for (function, expected_path) in [ + ("metadata_cache", "data/file.parquet"), + ("statistics_cache", "data/file.parquet"), + ("list_files_cache", "data"), + ] { + let df = ctx + .sql(&format!( + "SELECT object_store_url, path FROM {function}() ORDER BY object_store_url" + )) + .await?; + let all_columns = ctx.sql(&format!("SELECT * FROM {function}()")).await?; + let names = all_columns + .schema() + .fields() + .iter() + .map(|field| field.name().as_str()) + .collect::>(); + let expected = match function { + "metadata_cache" => vec![ + "path", + "file_modified", + "file_size_bytes", + "e_tag", + "version", + "metadata_size_bytes", + "hits", + "extra", + "object_store_url", + ], + "statistics_cache" => vec![ + "path", + "table", + "file_modified", + "file_size_bytes", + "e_tag", + "version", + "num_rows", + "num_columns", + "table_size_bytes", + "statistics_size_bytes", + "hits", + "object_store_url", + ], + "list_files_cache" => vec![ + "table", + "path", + "metadata_size_bytes", + "expires_in", + "metadata_list", + "hits", + "object_store_url", + ], + _ => unreachable!(), + }; + assert_eq!(names, expected, "{function}"); + let field = df + .schema() + .field_with_unqualified_name("object_store_url")?; + assert_eq!(field.data_type(), &DataType::Utf8, "{function}"); + assert!(!field.is_nullable(), "{function}"); + let results = df.collect().await?; + let expected = RecordBatch::try_from_iter([ + ( + "object_store_url", + Arc::new(StringArray::from(vec!["mem://first/", "mem://second/"])) + as ArrayRef, + ), + ( + "path", + Arc::new(StringArray::from(vec![expected_path, expected_path])), + ), + ])?; + assert_eq!( + batches_to_string(&results), + batches_to_string(&[expected]), + "{function}" + ); + } + Ok(()) + } + #[tokio::test] async fn test_metadata_cache() -> Result<(), DataFusionError> { let ctx = SessionContext::new(); diff --git a/datafusion/catalog-listing/src/options.rs b/datafusion/catalog-listing/src/options.rs index 44337e52a1e05..567fae89ba3a2 100644 --- a/datafusion/catalog-listing/src/options.rs +++ b/datafusion/catalog-listing/src/options.rs @@ -300,7 +300,14 @@ impl ListingOptions { .filter(|object_meta| object_meta.size > 0) .collect(); - let schema = self.format.infer_schema(state, &store, &files).await?; + let format = self + .format + .with_file_metadata_cache(Some(( + state.runtime_env().cache_manager.get_file_metadata_cache(), + Arc::new(table_path.object_store()), + ))) + .unwrap_or_else(|| Arc::clone(&self.format)); + let schema = format.infer_schema(state, &store, &files).await?; Ok(schema) } diff --git a/datafusion/catalog-listing/src/table.rs b/datafusion/catalog-listing/src/table.rs index c94ddcef22d6b..fb9eed85a9bc3 100644 --- a/datafusion/catalog-listing/src/table.rs +++ b/datafusion/catalog-listing/src/table.rs @@ -42,6 +42,7 @@ use datafusion_datasource::{ use datafusion_execution::cache::cache_manager::{ CachedFileMetadata, FileStatisticsCache, SchemaFingerprint, TableScopedPath, }; +use datafusion_execution::object_store::ObjectStoreUrl; use datafusion_expr::dml::InsertOp; use datafusion_expr::execution_props::ExecutionProps; use datafusion_expr::physical_planning_context::PhysicalPlanningContext; @@ -871,13 +872,16 @@ impl ListingTable { lfc.drop_table_entries(table_ref)?; } else { let table_prefix = table_path.prefix(); + let object_store_url = table_path.object_store(); let keys: Vec<_> = lfc .list_entries() .into_keys() .filter(|key| { key.table.is_none() - && (key.path.prefix_matches(table_prefix) - || table_prefix.prefix_matches(&key.path)) + && key.store_path.object_store_url.as_ref() + == &object_store_url + && (key.store_path.path.prefix_matches(table_prefix) + || table_prefix.prefix_matches(&key.store_path.path)) }) .collect(); for key in keys { @@ -941,6 +945,7 @@ impl ListingTable { &'a self, ctx: &'a dyn Session, store: &'a Arc, + object_store_url: &'a Arc, listing_time_filters: &'a [Expr], file_limit: Option, ) -> datafusion_common::Result<(FileGroup, bool)> { @@ -972,8 +977,13 @@ impl ListingTable { .map(|part_file| async { let part_file = part_file?; let (statistics, ordering) = if ctx.config().collect_statistics() { - self.do_collect_statistics_and_ordering(ctx, store, &part_file) - .await? + self.do_collect_statistics_and_ordering( + ctx, + store, + object_store_url, + &part_file, + ) + .await? } else { (Arc::new(Statistics::new_unknown(&self.file_schema)), None) }; @@ -1002,8 +1012,11 @@ impl ListingTable { ); } - let store = if let Some(url) = self.table_paths.first() { - ctx.runtime_env().object_store(url)? + let (store, object_store_url) = if let Some(url) = self.table_paths.first() { + ( + ctx.runtime_env().object_store(url)?, + Arc::new(url.object_store()), + ) } else { return Ok(ListFilesResult { file_groups: vec![], @@ -1013,7 +1026,7 @@ impl ListingTable { }); }; let (file_group, inexact_stats) = self - .collect_files_for_scan(ctx, &store, filters, limit) + .collect_files_for_scan(ctx, &store, &object_store_url, filters, limit) .await?; // Threshold: 0 = disabled, N > 0 = enabled when distinct_keys >= N @@ -1068,8 +1081,11 @@ impl ListingTable { ); } - let store = if let Some(url) = self.table_paths.first() { - ctx.runtime_env().object_store(url)? + let (store, object_store_url) = if let Some(url) = self.table_paths.first() { + ( + ctx.runtime_env().object_store(url)?, + Arc::new(url.object_store()), + ) } else { return Ok(ListFilesResult { file_groups: vec![], @@ -1078,8 +1094,9 @@ impl ListingTable { partition_split_points: vec![], }); }; - let (file_group, inexact_stats) = - self.collect_files_for_scan(ctx, &store, &[], None).await?; + let (file_group, inexact_stats) = self + .collect_files_for_scan(ctx, &store, &object_store_url, &[], None) + .await?; let mut file_groups = file_group.split_files(file_group_count); if !file_groups.is_empty() { file_groups.resize_with(file_group_count, || FileGroup::new(vec![])); @@ -1152,15 +1169,17 @@ impl ListingTable { &self, ctx: &dyn Session, store: &Arc, + object_store_url: &Arc, part_file: &PartitionedFile, ) -> datafusion_common::Result<(Arc, Option)> { - let path = TableScopedPath { - table: part_file.table_reference.clone(), - path: part_file.object_meta.location.clone(), - }; + let path = TableScopedPath::new( + part_file.table_reference.clone(), + Arc::clone(object_store_url), + part_file.object_meta.location.clone(), + ); let meta = &part_file.object_meta; - // Check cache first. The key stays `{table, path}` for cheap lookups; + // Check cache first. The key identifies the store, table and path; // the cached value carries the schema fingerprint to prevent reusing // stats computed under a different file schema. if let Some(cache) = &self.collected_statistics @@ -1172,9 +1191,15 @@ impl ListingTable { } // Cache miss or invalid: fetch both statistics and ordering in a single metadata read - let file_meta = self + let format = self .options .format + .with_file_metadata_cache(Some(( + ctx.runtime_env().cache_manager.get_file_metadata_cache(), + Arc::clone(object_store_url), + ))) + .unwrap_or_else(|| Arc::clone(&self.options.format)); + let file_meta = format .infer_stats_and_ordering(ctx, store, Arc::clone(&self.file_schema), meta) .await?; diff --git a/datafusion/core/src/datasource/file_format/parquet.rs b/datafusion/core/src/datasource/file_format/parquet.rs index c5685e1d2dfcb..66faec8b47a93 100644 --- a/datafusion/core/src/datasource/file_format/parquet.rs +++ b/datafusion/core/src/datasource/file_format/parquet.rs @@ -199,7 +199,10 @@ mod tests { let file_metadata_cache = ctx.runtime_env().cache_manager.get_file_metadata_cache(); let stats = DFParquetMetadata::new(&store, &meta[0]) - .with_file_metadata_cache(Some(Arc::clone(&file_metadata_cache))) + .with_file_metadata_cache(Some(( + Arc::clone(&file_metadata_cache), + (ObjectStoreUrl::local_filesystem()).into(), + ))) .fetch_statistics(&schema) .await?; @@ -210,7 +213,10 @@ mod tests { assert_eq!(c2_stats.null_count, Precision::Exact(3)); let stats = DFParquetMetadata::new(&store, &meta[1]) - .with_file_metadata_cache(Some(Arc::clone(&file_metadata_cache))) + .with_file_metadata_cache(Some(( + Arc::clone(&file_metadata_cache), + (ObjectStoreUrl::local_filesystem()).into(), + ))) .fetch_statistics(&schema) .await?; @@ -387,8 +393,10 @@ mod tests { df_meta.fetch_metadata().await?; assert_eq!(store.request_count(), 2); - let df_meta = - df_meta.with_file_metadata_cache(Some(Arc::clone(&file_metadata_cache))); + let df_meta = df_meta.with_file_metadata_cache(Some(( + Arc::clone(&file_metadata_cache), + (ObjectStoreUrl::local_filesystem()).into(), + ))); // Increases by 3 because cache has no entries yet df_meta.fetch_metadata().await?; @@ -409,7 +417,12 @@ mod tests { }; let format = ParquetFormat::default() .with_metadata_size_hint(Some(9)) - .with_force_view_types(force_views); + .with_force_view_types(force_views) + .with_file_metadata_cache(Some(( + Arc::clone(&file_metadata_cache), + Arc::new(ObjectStoreUrl::local_filesystem()), + ))) + .unwrap(); // Increase by 3, partial cache being used. let _schema = format.infer_schema(&ctx, &store.upcast(), &meta).await?; assert_eq!(store.request_count(), 10); @@ -418,8 +431,10 @@ mod tests { assert_eq!(store.request_count(), 10); // No increase, cache being used - let df_meta = - df_meta.with_file_metadata_cache(Some(Arc::clone(&file_metadata_cache))); + let df_meta = df_meta.with_file_metadata_cache(Some(( + Arc::clone(&file_metadata_cache), + (ObjectStoreUrl::local_filesystem()).into(), + ))); let stats = df_meta.fetch_statistics(&schema).await?; assert_eq!(store.request_count(), 10); @@ -446,8 +461,10 @@ mod tests { let ctx = session.state(); let file_metadata_cache = ctx.runtime_env().cache_manager.get_file_metadata_cache(); - let df_meta = - df_meta.with_file_metadata_cache(Some(Arc::clone(&file_metadata_cache))); + let df_meta = df_meta.with_file_metadata_cache(Some(( + Arc::clone(&file_metadata_cache), + (ObjectStoreUrl::local_filesystem()).into(), + ))); // Increases by 1 because cache has no entries yet and new session context df_meta.fetch_metadata().await?; assert_eq!(store.request_count(), 2); @@ -463,7 +480,12 @@ mod tests { let format = ParquetFormat::default() .with_metadata_size_hint(Some(size_hint)) - .with_force_view_types(force_views); + .with_force_view_types(force_views) + .with_file_metadata_cache(Some(( + Arc::clone(&file_metadata_cache), + Arc::new(ObjectStoreUrl::local_filesystem()), + ))) + .unwrap(); // Increase by 1, partial cache being used. let _schema = format.infer_schema(&ctx, &store.upcast(), &meta).await?; assert_eq!(store.request_count(), 4); @@ -471,8 +493,10 @@ mod tests { let schema = format.infer_schema(&ctx, &store.upcast(), &meta).await?; assert_eq!(store.request_count(), 4); // No increase, cache being used - let df_meta = - df_meta.with_file_metadata_cache(Some(Arc::clone(&file_metadata_cache))); + let df_meta = df_meta.with_file_metadata_cache(Some(( + Arc::clone(&file_metadata_cache), + (ObjectStoreUrl::local_filesystem()).into(), + ))); let stats = df_meta.fetch_statistics(&schema).await?; assert_eq!(store.request_count(), 4); @@ -495,8 +519,10 @@ mod tests { assert_eq!(store.request_count(), 1); // No increase because cache has an entry - let df_meta = - df_meta.with_file_metadata_cache(Some(Arc::clone(&file_metadata_cache))); + let df_meta = df_meta.with_file_metadata_cache(Some(( + Arc::clone(&file_metadata_cache), + (ObjectStoreUrl::local_filesystem()).into(), + ))); df_meta.fetch_metadata().await?; assert_eq!(store.request_count(), 1); @@ -543,7 +569,13 @@ mod tests { let state = SessionContext::new().state(); // Make metadata size hint None to keep original behavior - let format = ParquetFormat::default().with_metadata_size_hint(None); + let format = ParquetFormat::default() + .with_metadata_size_hint(None) + .with_file_metadata_cache(Some(( + state.runtime_env().cache_manager.get_file_metadata_cache(), + Arc::new(ObjectStoreUrl::local_filesystem()), + ))) + .unwrap(); let _schema = format.infer_schema(&state, &store.upcast(), &files).await?; assert_eq!(store.request_count(), 3); // No increase, cache being used. @@ -554,7 +586,10 @@ mod tests { let file_metadata_cache = state.runtime_env().cache_manager.get_file_metadata_cache(); let stats = DFParquetMetadata::new(store.as_ref(), &files[0]) - .with_file_metadata_cache(Some(Arc::clone(&file_metadata_cache))) + .with_file_metadata_cache(Some(( + Arc::clone(&file_metadata_cache), + (ObjectStoreUrl::local_filesystem()).into(), + ))) .fetch_statistics(&schema) .await?; assert_eq!(stats.num_rows, Precision::Exact(4)); @@ -619,7 +654,12 @@ mod tests { state = set_view_state(state, force_views); let format = ParquetFormat::default() .with_force_view_types(force_views) - .with_metadata_size_hint(None); + .with_metadata_size_hint(None) + .with_file_metadata_cache(Some(( + state.runtime_env().cache_manager.get_file_metadata_cache(), + Arc::new(ObjectStoreUrl::local_filesystem()), + ))) + .unwrap(); let schema = format.infer_schema(&state, &store.upcast(), &files).await?; assert_eq!(store.request_count(), 6); @@ -634,7 +674,10 @@ mod tests { let file_metadata_cache = state.runtime_env().cache_manager.get_file_metadata_cache(); let stats = DFParquetMetadata::new(store.as_ref(), &files[0]) - .with_file_metadata_cache(Some(Arc::clone(&file_metadata_cache))) + .with_file_metadata_cache(Some(( + Arc::clone(&file_metadata_cache), + (ObjectStoreUrl::local_filesystem()).into(), + ))) .fetch_statistics(&schema) .await?; assert_eq!(store.request_count(), 6); @@ -663,7 +706,10 @@ mod tests { // No increase in request count because cache is not empty let stats = DFParquetMetadata::new(store.as_ref(), &files[1]) - .with_file_metadata_cache(Some(Arc::clone(&file_metadata_cache))) + .with_file_metadata_cache(Some(( + Arc::clone(&file_metadata_cache), + (ObjectStoreUrl::local_filesystem()).into(), + ))) .fetch_statistics(&schema) .await?; assert_eq!(store.request_count(), 6); diff --git a/datafusion/core/tests/datasource/cache_object_identity.rs b/datafusion/core/tests/datasource/cache_object_identity.rs new file mode 100644 index 0000000000000..688a9ccfeb65f --- /dev/null +++ b/datafusion/core/tests/datasource/cache_object_identity.rs @@ -0,0 +1,513 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use arrow::array::{ArrayRef, Int32Array, RecordBatch}; +use async_trait::async_trait; +use bytes::Bytes; +use datafusion::datasource::file_format::{FileFormat, parquet::ParquetFormat}; +use datafusion::prelude::{ParquetReadOptions, SessionContext, col, lit}; +use datafusion_common::{ScalarValue, assert_batches_eq, stats::Precision}; +use datafusion_execution::cache::{StoreScopedPath, TableScopedPath}; +use datafusion_execution::object_store::ObjectStoreUrl; +use datafusion_functions_aggregate::expr_fn::{count, max, min}; +use futures::stream::BoxStream; +use futures::{StreamExt, TryStreamExt}; +use object_store::memory::InMemory; +use object_store::path::Path; +use object_store::{ + CopyOptions, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta, + ObjectStore, ObjectStoreExt, PutMultipartOptions, PutOptions, PutPayload, PutResult, +}; +use parquet::arrow::ArrowWriter; +use parquet::basic::{Compression, Encoding}; +use parquet::file::properties::WriterProperties; +use std::fmt::{Display, Formatter}; +use std::sync::Arc; +use url::Url; + +fn parquet_bytes(values: Vec) -> Bytes { + let values: ArrayRef = Arc::new(Int32Array::from(values)); + let batch = RecordBatch::try_from_iter([("value", values)]).unwrap(); + let props = WriterProperties::builder() + .set_dictionary_enabled(false) + .set_encoding(Encoding::PLAIN) + .set_compression(Compression::UNCOMPRESSED) + .build(); + let mut bytes = vec![]; + let mut writer = + ArrowWriter::try_new(&mut bytes, batch.schema(), Some(props)).unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + bytes.into() +} + +async fn aggregate_file(ctx: &SessionContext, url: &str) -> Vec { + ctx.read_parquet(url, ParquetReadOptions::default()) + .await + .unwrap() + .aggregate( + vec![], + vec![ + min(col("value")).alias("min"), + max(col("value")).alias("max"), + count(col("value")).alias("count"), + ], + ) + .unwrap() + .collect() + .await + .unwrap() +} + +#[tokio::test] +async fn parquet_caches_separate_object_stores() { + let first = Arc::new(FixedTimeStore::default()); + let second = Arc::new(FixedTimeStore::default()); + let path = Path::from("data/file.parquet"); + let first_bytes = parquet_bytes(vec![1, 2, 3]); + let second_bytes = parquet_bytes(vec![7, 8, 9]); + assert_eq!(first_bytes.len(), second_bytes.len()); + first.put(&path, first_bytes.into()).await.unwrap(); + second.put(&path, second_bytes.into()).await.unwrap(); + // Every ObjectMeta field, including ETag, is identical across the stores. + assert_eq!( + first.head(&path).await.unwrap(), + second.head(&path).await.unwrap() + ); + + let ctx = SessionContext::new(); + ctx.runtime_env() + .register_object_store(&Url::parse("mem://first").unwrap(), first); + ctx.runtime_env() + .register_object_store(&Url::parse("mem://second").unwrap(), second); + let first_result = aggregate_file(&ctx, "mem://first/data/").await; + assert_batches_eq!( + [ + "+-----+-----+-------+", + "| min | max | count |", + "+-----+-----+-------+", + "| 1 | 3 | 3 |", + "+-----+-----+-------+", + ], + &first_result + ); + let second_result = aggregate_file(&ctx, "mem://second/data/").await; + assert_batches_eq!( + [ + "+-----+-----+-------+", + "| min | max | count |", + "+-----+-----+-------+", + "| 7 | 9 | 3 |", + "+-----+-----+-------+", + ], + &second_result + ); + + // Execute scans as well as metadata-backed aggregates, including repeated + // reads through the Parquet reader's metadata cache. + for (url, value) in [("mem://first/data/", 2), ("mem://second/data/", 8)] { + for _ in 0..2 { + let results = ctx + .read_parquet(url, ParquetReadOptions::default()) + .await + .unwrap() + .filter(col("value").eq(lit(value))) + .unwrap() + .collect() + .await + .unwrap(); + let values = results + .iter() + .flat_map(|batch| { + batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .iter() + }) + .collect::>(); + assert_eq!(values, vec![Some(value)], "{url}"); + } + } + + let caches = &ctx.runtime_env().cache_manager; + let metadata = caches.get_file_metadata_cache(); + let statistics = caches.get_file_statistic_cache().unwrap(); + let listings = caches.get_list_files_cache().unwrap(); + for url in ["mem://first", "mem://second"] { + let object_store_url = ObjectStoreUrl::parse(url).unwrap(); + assert!( + metadata + .get(&StoreScopedPath::new( + object_store_url.clone(), + path.clone() + )) + .is_some() + ); + assert!( + statistics + .get(&TableScopedPath { + table: None, + store_path: StoreScopedPath::new( + object_store_url.clone(), + path.clone() + ), + }) + .is_some() + ); + assert!( + listings + .get(&TableScopedPath { + table: None, + store_path: StoreScopedPath::new( + object_store_url, + Path::from("data") + ), + }) + .is_some() + ); + } + assert_eq!(metadata.len(), 2); + assert_eq!(statistics.len(), 2); + assert_eq!(listings.len(), 2); +} + +#[tokio::test] +async fn parquet_caches_reregistered_object_store() { + for deregister in [false, true] { + let first = Arc::new(FixedTimeStore::default()); + let second = Arc::new(FixedTimeStore::default()); + let path = Path::from("data.parquet"); + let first_bytes = parquet_bytes(vec![1, 2, 3]); + let second_bytes = parquet_bytes(vec![7, 8, 9]); + assert_eq!(first_bytes.len(), second_bytes.len()); + first.put(&path, first_bytes.into()).await.unwrap(); + second.put(&path, second_bytes.into()).await.unwrap(); + assert_eq!( + first.head(&path).await.unwrap(), + second.head(&path).await.unwrap() + ); + + let ctx = SessionContext::new(); + let url = Url::parse("mem://reused").unwrap(); + ctx.register_object_store(&url, first); + let first_result = aggregate_file(&ctx, "mem://reused/data.parquet").await; + assert_batches_eq!( + [ + "+-----+-----+-------+", + "| min | max | count |", + "+-----+-----+-------+", + "| 1 | 3 | 3 |", + "+-----+-----+-------+", + ], + &first_result + ); + if deregister { + ctx.deregister_object_store(&url).unwrap(); + } + // Registration ignores paths and credentials: invalidate the same scope. + let registration_url = Url::parse("mem://user:password@reused/ignored").unwrap(); + ctx.register_object_store(®istration_url, second); + let second_result = aggregate_file(&ctx, "mem://reused/data.parquet").await; + assert_batches_eq!( + [ + "+-----+-----+-------+", + "| min | max | count |", + "+-----+-----+-------+", + "| 7 | 9 | 3 |", + "+-----+-----+-------+", + ], + &second_result + ); + } +} + +#[tokio::test] +async fn reregistering_store_invalidates_only_its_caches() { + let first = Arc::new(FixedTimeStore::default()); + let second = Arc::new(FixedTimeStore::default()); + let other = Arc::new(FixedTimeStore::default()); + let path = Path::from("data/file.parquet"); + first + .put(&path, parquet_bytes(vec![1, 2, 3]).into()) + .await + .unwrap(); + first + .put( + &Path::from("data/removed.parquet"), + parquet_bytes(vec![0]).into(), + ) + .await + .unwrap(); + second + .put(&path, parquet_bytes(vec![7, 8, 9]).into()) + .await + .unwrap(); + other + .put(&path, parquet_bytes(vec![4, 5, 6]).into()) + .await + .unwrap(); + let ctx = SessionContext::new(); + let url = Url::parse("mem://reused").unwrap(); + ctx.register_object_store(&url, first); + ctx.register_object_store(&Url::parse("mem://other").unwrap(), other); + aggregate_file(&ctx, "mem://reused/data/").await; + aggregate_file(&ctx, "mem://other/data/").await; + ctx.register_parquet("named", "mem://reused/data/", ParquetReadOptions::default()) + .await + .unwrap(); + ctx.sql("SELECT min(value), max(value), count(*) FROM named") + .await + .unwrap() + .collect() + .await + .unwrap(); + + let caches = &ctx.runtime_env().cache_manager; + let metadata = caches.get_file_metadata_cache(); + let statistics = caches.get_file_statistic_cache().unwrap(); + let listings = caches.get_list_files_cache().unwrap(); + assert!( + statistics + .list_entries() + .keys() + .any(|key| key.table.is_some()) + ); + assert!( + listings + .list_entries() + .keys() + .any(|key| key.table.is_some()) + ); + ctx.register_object_store(&url, second); + assert_eq!(metadata.len(), 1); + assert_eq!(statistics.len(), 1); + assert_eq!(listings.len(), 1); + assert!( + metadata + .list_entries() + .keys() + .all(|key| key.object_store_url.as_str() == "mem://other/") + ); + assert!( + statistics.list_entries().keys().all(|key| key + .store_path + .object_store_url + .as_str() + == "mem://other/") + ); + assert!( + listings.list_entries().keys().all(|key| key + .store_path + .object_store_url + .as_str() + == "mem://other/") + ); + + // A stale listing would include removed.parquet, which no longer exists. + let result = aggregate_file(&ctx, "mem://reused/data/").await; + assert_batches_eq!( + [ + "+-----+-----+-------+", + "| min | max | count |", + "+-----+-----+-------+", + "| 7 | 9 | 3 |", + "+-----+-----+-------+", + ], + &result + ); +} + +#[tokio::test] +async fn unscoped_format_inference_reads_each_store() { + let first: Arc = Arc::new(FixedTimeStore::default()); + let second: Arc = Arc::new(FixedTimeStore::default()); + let path = Path::from("data.parquet"); + first + .put(&path, parquet_bytes(vec![1, 2, 3]).into()) + .await + .unwrap(); + second + .put(&path, parquet_bytes(vec![7, 8, 9]).into()) + .await + .unwrap(); + let meta = first.head(&path).await.unwrap(); + assert_eq!(meta, second.head(&path).await.unwrap()); + let state = SessionContext::new().state(); + let format = ParquetFormat::default(); + let schema = format + .infer_schema(&state, &first, std::slice::from_ref(&meta)) + .await + .unwrap(); + for (store, expected_min, expected_max) in [(first, 1, 3), (second, 7, 9)] { + let stats = format + .infer_stats(&state, &store, Arc::clone(&schema), &meta) + .await + .unwrap(); + assert_eq!( + stats.column_statistics[0].min_value, + Precision::Exact(ScalarValue::Int32(Some(expected_min))) + ); + assert_eq!( + stats.column_statistics[0].max_value, + Precision::Exact(ScalarValue::Int32(Some(expected_max))) + ); + } +} + +#[tokio::test] +async fn parquet_caches_refresh_changed_etag() { + let store = Arc::new(FixedTimeStore::default()); + let path = Path::from("data.parquet"); + let original = parquet_bytes(vec![1, 2, 3]); + let replacement = parquet_bytes(vec![7, 8, 9]); + assert_eq!(original.len(), replacement.len()); + store.put(&path, original.into()).await.unwrap(); + let before = store.head(&path).await.unwrap(); + let ctx = SessionContext::new(); + ctx.runtime_env() + .register_object_store(&Url::parse("mem://overwrite").unwrap(), store.clone()); + // A single-file URL obtains fresh metadata without a cached directory listing. + let original_result = aggregate_file(&ctx, "mem://overwrite/data.parquet").await; + assert_batches_eq!( + [ + "+-----+-----+-------+", + "| min | max | count |", + "+-----+-----+-------+", + "| 1 | 3 | 3 |", + "+-----+-----+-------+", + ], + &original_result + ); + + store.put(&path, replacement.into()).await.unwrap(); + let after = store.head(&path).await.unwrap(); + assert_eq!(before.size, after.size); + assert_eq!(before.last_modified, after.last_modified); + assert_ne!(before.e_tag, after.e_tag); + let replacement_result = aggregate_file(&ctx, "mem://overwrite/data.parquet").await; + assert_batches_eq!( + [ + "+-----+-----+-------+", + "| min | max | count |", + "+-----+-----+-------+", + "| 7 | 9 | 3 |", + "+-----+-----+-------+", + ], + &replacement_result + ); + let object_store_url = ObjectStoreUrl::parse("mem://overwrite").unwrap(); + let caches = &ctx.runtime_env().cache_manager; + let metadata = caches + .get_file_metadata_cache() + .get(&StoreScopedPath::new( + object_store_url.clone(), + path.clone(), + )) + .unwrap(); + let statistics = caches + .get_file_statistic_cache() + .unwrap() + .get(&TableScopedPath { + table: None, + store_path: StoreScopedPath::new(object_store_url, path), + }) + .unwrap(); + assert_eq!(metadata.meta.e_tag, after.e_tag); + assert_eq!(statistics.meta.e_tag, after.e_tag); +} + +/// Keeps modification times identical so the tests isolate store identity and ETag. +#[derive(Debug, Default)] +struct FixedTimeStore(InMemory); + +impl Display for FixedTimeStore { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "FixedTimeStore") + } +} + +#[async_trait] +impl ObjectStore for FixedTimeStore { + async fn put_opts( + &self, + location: &Path, + payload: PutPayload, + opts: PutOptions, + ) -> object_store::Result { + self.0.put_opts(location, payload, opts).await + } + + async fn put_multipart_opts( + &self, + location: &Path, + opts: PutMultipartOptions, + ) -> object_store::Result> { + self.0.put_multipart_opts(location, opts).await + } + + async fn get_opts( + &self, + location: &Path, + options: GetOptions, + ) -> object_store::Result { + let mut result = self.0.get_opts(location, options).await?; + result.meta.last_modified = chrono::DateTime::UNIX_EPOCH; + Ok(result) + } + + fn list( + &self, + prefix: Option<&Path>, + ) -> BoxStream<'static, object_store::Result> { + self.0 + .list(prefix) + .map_ok(|mut meta| { + meta.last_modified = chrono::DateTime::UNIX_EPOCH; + meta + }) + .boxed() + } + + async fn list_with_delimiter( + &self, + prefix: Option<&Path>, + ) -> object_store::Result { + let mut result = self.0.list_with_delimiter(prefix).await?; + for meta in &mut result.objects { + meta.last_modified = chrono::DateTime::UNIX_EPOCH; + } + Ok(result) + } + + fn delete_stream( + &self, + locations: BoxStream<'static, object_store::Result>, + ) -> BoxStream<'static, object_store::Result> { + self.0.delete_stream(locations) + } + + async fn copy_opts( + &self, + from: &Path, + to: &Path, + options: CopyOptions, + ) -> object_store::Result<()> { + self.0.copy_opts(from, to, options).await + } +} diff --git a/datafusion/core/tests/datasource/mod.rs b/datafusion/core/tests/datasource/mod.rs index 3785aa0766182..109f19aaa02d1 100644 --- a/datafusion/core/tests/datasource/mod.rs +++ b/datafusion/core/tests/datasource/mod.rs @@ -19,6 +19,7 @@ //! //! Note tests for the Parquet format are in `parquet_integration` binary +mod cache_object_identity; // Include tests in csv module mod csv; mod object_store_access; diff --git a/datafusion/core/tests/datasource/object_store_access.rs b/datafusion/core/tests/datasource/object_store_access.rs index 215725cf04961..9dd56d4614635 100644 --- a/datafusion/core/tests/datasource/object_store_access.rs +++ b/datafusion/core/tests/datasource/object_store_access.rs @@ -36,7 +36,8 @@ use datafusion_common::assert_batches_eq; use datafusion_datasource::ListingTableUrl; use datafusion_datasource_csv::CsvFormat; use datafusion_datasource_json::JsonFormat; -use datafusion_execution::cache::TableScopedPath; +use datafusion_execution::cache::{StoreScopedPath, TableScopedPath}; +use datafusion_execution::object_store::ObjectStoreUrl; use futures::stream::BoxStream; use insta::assert_snapshot; use object_store::memory::InMemory; @@ -219,6 +220,13 @@ async fn insert_invalidates_overlapping_unscoped_listings() { let ctx = SessionContext::new(); ctx.runtime_env() .register_object_store(&Url::parse("mem://").unwrap(), store); + let other_store = Arc::new(InMemory::new()); + other_store + .put(&Path::from("table/data.csv"), "4\n".into()) + .await + .unwrap(); + ctx.runtime_env() + .register_object_store(&Url::parse("mem://other").unwrap(), other_store); ctx.sql("SET datafusion.execution.listing_table_ignore_subdirectory = false") .await .unwrap() @@ -236,15 +244,16 @@ async fn insert_invalidates_overlapping_unscoped_listings() { .get_list_files_cache() .unwrap(); let tables = [ - ("root_table", "table"), - ("child_table", "table/region=US"), - ("descendant_table", "table/region=US/q1"), - ("sibling_table", "table/region=EU"), + ("root_table", "mem://", "table"), + ("child_table", "mem://", "table/region=US"), + ("descendant_table", "mem://", "table/region=US/q1"), + ("sibling_table", "mem://", "table/region=EU"), + ("other_store_table", "mem://other", "table"), ]; - for (name, path) in tables { + for (name, store_url, path) in tables { // Keep the URLs unscoped so overlapping tables share path-based cache entries. - let url = ListingTableUrl::parse(format!("mem:///{path}/")).unwrap(); + let url = ListingTableUrl::parse(format!("{store_url}/{path}/")).unwrap(); let options = ListingOptions::new(Arc::new(CsvFormat::default().with_has_header(false))) .with_file_extension(".csv"); @@ -261,7 +270,10 @@ async fn insert_invalidates_overlapping_unscoped_listings() { .unwrap(); let key = TableScopedPath { table: None, - path: Path::from(path), + store_path: StoreScopedPath::new( + ObjectStoreUrl::parse(store_url).unwrap(), + Path::from(path), + ), }; assert!(cache.get(&key).is_some()); } @@ -273,13 +285,20 @@ async fn insert_invalidates_overlapping_unscoped_listings() { .await .unwrap(); - for (name, path) in tables { + for (name, store_url, path) in tables { let key = TableScopedPath { table: None, - path: Path::from(path), + store_path: StoreScopedPath::new( + ObjectStoreUrl::parse(store_url).unwrap(), + Path::from(path), + ), }; let cached = cache.get(&key); - assert_eq!(cached.is_some(), name == "sibling_table", "{name}"); + assert_eq!( + cached.is_some(), + matches!(name, "sibling_table" | "other_store_table"), + "{name}" + ); } let batches = ctx diff --git a/datafusion/datasource-parquet/src/file_format.rs b/datafusion/datasource-parquet/src/file_format.rs index aa8e82c9707be..c9c66144ae220 100644 --- a/datafusion/datasource-parquet/src/file_format.rs +++ b/datafusion/datasource-parquet/src/file_format.rs @@ -55,6 +55,8 @@ use datafusion_datasource::file::FileSource; use datafusion_datasource::file_scan_config::{FileScanConfig, FileScanConfigBuilder}; use datafusion_datasource::sink::DataSinkExec; use datafusion_datasource::write::get_writer_schema; +use datafusion_execution::cache::cache_manager::FileMetadataCache; +use datafusion_execution::object_store::ObjectStoreUrl; use datafusion_expr::dml::InsertOp; use datafusion_physical_expr_common::sort_expr::{LexOrdering, LexRequirement}; use datafusion_physical_plan::ExecutionPlan; @@ -144,6 +146,7 @@ impl Debug for ParquetFormatFactory { #[derive(Debug, Default)] pub struct ParquetFormat { options: TableParquetOptions, + metadata_cache: Option<(Arc, Arc)>, } impl ParquetFormat { @@ -403,6 +406,16 @@ async fn get_file_decryption_properties( #[async_trait] impl FileFormat for ParquetFormat { + fn with_file_metadata_cache( + &self, + metadata_cache: Option<(Arc, Arc)>, + ) -> Option> { + Some(Arc::new(Self { + options: self.options.clone(), + metadata_cache, + })) + } + fn get_ext(&self) -> String { ParquetFormatFactory::new().get_ext() } @@ -436,9 +449,6 @@ impl FileFormat for ParquetFormat { .as_ref() .map(|tz| Arc::::from(tz.as_str())); - let file_metadata_cache = - state.runtime_env().cache_manager.get_file_metadata_cache(); - let mut schemas: Vec<_> = futures::stream::iter(objects) .map(|object| async { let file_decryption_properties = get_file_decryption_properties( @@ -450,7 +460,7 @@ impl FileFormat for ParquetFormat { let meta = DFParquetMetadata::new(store.as_ref(), object) .with_metadata_size_hint(self.metadata_size_hint()) .with_decryption_properties(file_decryption_properties) - .with_file_metadata_cache(Some(Arc::clone(&file_metadata_cache))) + .with_file_metadata_cache(self.metadata_cache.clone()) .with_coerce_int96(coerce_int96) .with_coerce_int96_tz(coerce_int96_tz.clone()) .with_enable_rle_to_dictionary( @@ -532,12 +542,10 @@ impl FileFormat for ParquetFormat { let file_decryption_properties = get_file_decryption_properties(state, &self.options, &object.location) .await?; - let file_metadata_cache = - state.runtime_env().cache_manager.get_file_metadata_cache(); DFParquetMetadata::new(store, object) .with_metadata_size_hint(self.metadata_size_hint()) .with_decryption_properties(file_decryption_properties) - .with_file_metadata_cache(Some(file_metadata_cache)) + .with_file_metadata_cache(self.metadata_cache.clone()) .fetch_statistics(&table_schema) .await } @@ -552,12 +560,10 @@ impl FileFormat for ParquetFormat { let file_decryption_properties = get_file_decryption_properties(state, &self.options, &object.location) .await?; - let file_metadata_cache = - state.runtime_env().cache_manager.get_file_metadata_cache(); let metadata = DFParquetMetadata::new(store, object) .with_metadata_size_hint(self.metadata_size_hint()) .with_decryption_properties(file_decryption_properties) - .with_file_metadata_cache(Some(file_metadata_cache)) + .with_file_metadata_cache(self.metadata_cache.clone()) .fetch_metadata() .await?; crate::metadata::ordering_from_parquet_metadata(&metadata, &table_schema) @@ -573,12 +579,10 @@ impl FileFormat for ParquetFormat { let file_decryption_properties = get_file_decryption_properties(state, &self.options, &object.location) .await?; - let file_metadata_cache = - state.runtime_env().cache_manager.get_file_metadata_cache(); let metadata = DFParquetMetadata::new(store, object) .with_metadata_size_hint(self.metadata_size_hint()) .with_decryption_properties(file_decryption_properties) - .with_file_metadata_cache(Some(file_metadata_cache)) + .with_file_metadata_cache(self.metadata_cache.clone()) .fetch_metadata() .await?; let statistics = DFParquetMetadata::statistics_from_parquet_metadata( @@ -616,8 +620,11 @@ impl FileFormat for ParquetFormat { let store = state .runtime_env() .object_store(conf.object_store_url.clone())?; - let cached_parquet_read_factory = - Arc::new(CachedParquetFileReaderFactory::new(store, metadata_cache)); + let cached_parquet_read_factory = Arc::new(CachedParquetFileReaderFactory::new( + store, + conf.object_store_url.clone(), + metadata_cache, + )); source = source.with_parquet_file_reader_factory(cached_parquet_read_factory); if let Some(metadata_size_hint) = metadata_size_hint { diff --git a/datafusion/datasource-parquet/src/metadata.rs b/datafusion/datasource-parquet/src/metadata.rs index 0cfcb980a788f..46c4de62979cf 100644 --- a/datafusion/datasource-parquet/src/metadata.rs +++ b/datafusion/datasource-parquet/src/metadata.rs @@ -32,8 +32,9 @@ use datafusion_common::{ internal_datafusion_err, }; use datafusion_execution::cache::cache_manager::{ - CachedFileMetadataEntry, FileMetadata, FileMetadataCache, + CachedFileMetadataEntry, FileMetadata, FileMetadataCache, StoreScopedPath, }; +use datafusion_execution::object_store::ObjectStoreUrl; use datafusion_functions_aggregate_common::min_max::{MaxAccumulator, MinAccumulator}; use datafusion_physical_expr::expressions::Column; use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr}; @@ -140,8 +141,8 @@ pub struct DFParquetMetadata<'a> { /// [`ParquetMetaDataReader::with_decryption_properties`]. decryption_properties: Option>, /// Optional cache of previously fetched [`ParquetMetaData`], keyed by - /// file location. - file_metadata_cache: Option>, + /// object store URL and file location. + file_metadata_cache: Option<(Arc, StoreScopedPath)>, /// Policy controlling whether the Parquet page index (column and offset /// indexes) is fetched, mirroring /// [`ParquetMetaDataReader::with_page_index_policy`]. @@ -209,11 +210,18 @@ impl<'a> DFParquetMetadata<'a> { /// Set an optional [`FileMetadataCache`] used to avoid re-fetching /// [`ParquetMetaData`] for files that have already been read. + /// The paired URL identifies the store supplying this file, so equal + /// paths in different stores cannot reuse each other's metadata. pub fn with_file_metadata_cache( mut self, - file_metadata_cache: Option>, + file_metadata_cache: Option<(Arc, Arc)>, ) -> Self { - self.file_metadata_cache = file_metadata_cache; + self.file_metadata_cache = file_metadata_cache.map(|(cache, url)| { + ( + cache, + StoreScopedPath::new(url, self.object_meta.location.clone()), + ) + }); self } @@ -280,8 +288,8 @@ impl<'a> DFParquetMetadata<'a> { let page_index_policy = self.effective_page_index_policy(cache_metadata); if cache_metadata - && let Some(file_metadata_cache) = self.file_metadata_cache.as_ref() - && let Some(cached) = file_metadata_cache.get(&self.object_meta.location) + && let Some((file_metadata_cache, key)) = self.file_metadata_cache.as_ref() + && let Some(cached) = file_metadata_cache.get(key) && cached.is_valid_for(self.object_meta) && let Some(cached_parquet) = cached .file_metadata @@ -339,14 +347,14 @@ impl<'a> DFParquetMetadata<'a> { } /// Store `metadata` in the configured [`FileMetadataCache`], keyed by - /// the file's location. + /// the object store URL and file's location. /// /// This is a no-op unless a cache has been configured via /// [`Self::with_file_metadata_cache`]. fn cache_metadata(&self, metadata: Arc) -> Result<()> { - if let Some(file_metadata_cache) = &self.file_metadata_cache { + if let Some((file_metadata_cache, key)) = &self.file_metadata_cache { file_metadata_cache.put( - &self.object_meta.location, + key, CachedFileMetadataEntry::new( self.object_meta.clone(), Arc::new(CachedParquetMetaData::new(metadata)), @@ -1267,6 +1275,114 @@ mod tests { use arrow::array::Int32Array; use arrow::compute::SortOptions; use arrow::datatypes::Field; + use arrow::record_batch::RecordBatch; + use datafusion_execution::cache::default_cache::DefaultCache; + use object_store::ObjectStoreExt; + use object_store::memory::InMemory; + use parquet::arrow::ArrowWriter; + + async fn write_cached_test_file(store: &InMemory, value: i32) -> ObjectMeta { + let batch = RecordBatch::try_from_iter(vec![( + "value", + Arc::new(Int32Array::from(vec![value])) as ArrayRef, + )]) + .unwrap(); + let mut bytes = Vec::new(); + let mut writer = ArrowWriter::try_new(&mut bytes, batch.schema(), None).unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + let path = Path::from("data.parquet"); + store.put(&path, bytes.into()).await.unwrap(); + store.head(&path).await.unwrap() + } + + #[tokio::test] + async fn metadata_cache_invalidates_changed_object_identity() { + for use_version in [false, true] { + let store = InMemory::new(); + let cache: Arc = Arc::new(DefaultCache::new(1024 * 1024)); + let url = ObjectStoreUrl::parse("s3://bucket").unwrap(); + let mut original = write_cached_test_file(&store, 100).await; + if use_version { + original.e_tag = None; + original.version = Some("v1".into()); + } + let original_reader = DFParquetMetadata::new(&store, &original) + .with_file_metadata_cache(Some(( + Arc::clone(&cache), + (url.clone()).into(), + ))) + .with_page_index_policy(Some(PageIndexPolicy::Skip)); + let original_metadata = original_reader.fetch_metadata().await.unwrap(); + assert!(Arc::ptr_eq( + &original_metadata, + &original_reader.fetch_metadata().await.unwrap() + )); + + let mut overwritten = write_cached_test_file(&store, 200).await; + assert_eq!(original.size, overwritten.size); + overwritten.last_modified = original.last_modified; + if use_version { + overwritten.e_tag = None; + overwritten.version = Some("v2".into()); + } else { + assert_ne!(original.e_tag, overwritten.e_tag); + } + let reader = DFParquetMetadata::new(&store, &overwritten) + .with_file_metadata_cache(Some((Arc::clone(&cache), (url).into()))) + .with_page_index_policy(Some(PageIndexPolicy::Skip)); + let metadata = reader.fetch_metadata().await.unwrap(); + assert!(!Arc::ptr_eq(&original_metadata, &metadata)); + let ParquetStatistics::Int32(stats) = + metadata.row_group(0).column(0).statistics().unwrap() + else { + panic!("expected Int32 statistics"); + }; + assert_eq!(stats.min_opt(), Some(&200)); + assert_eq!(stats.max_opt(), Some(&200)); + assert!(Arc::ptr_eq( + &metadata, + &reader.fetch_metadata().await.unwrap() + )); + assert_eq!(cache.len(), 1); + } + } + + #[tokio::test] + async fn metadata_cache_separates_object_stores() { + let first_store = InMemory::new(); + let second_store = InMemory::new(); + let meta = write_cached_test_file(&first_store, 100).await; + let second_meta = write_cached_test_file(&second_store, 200).await; + assert_eq!(meta.size, second_meta.size); + let cache: Arc = Arc::new(DefaultCache::new(1024 * 1024)); + // Use identical metadata deliberately: store identity must distinguish + // the files even when their paths and all validation fields coincide. + for _ in 0..2 { + for (store, url, expected) in [ + (&first_store, "s3://first", 100), + (&second_store, "s3://second", 200), + ] { + let metadata = DFParquetMetadata::new(store, &meta) + .with_file_metadata_cache(Some(( + Arc::clone(&cache), + (ObjectStoreUrl::parse(url).unwrap()).into(), + ))) + .with_page_index_policy(Some(PageIndexPolicy::Skip)) + .fetch_metadata() + .await + .unwrap(); + let ParquetStatistics::Int32(stats) = + metadata.row_group(0).column(0).statistics().unwrap() + else { + panic!("expected Int32 statistics"); + }; + assert_eq!(stats.min_opt(), Some(&expected)); + assert_eq!(stats.max_opt(), Some(&expected)); + } + } + assert_eq!(cache.len(), 2); + } #[test] fn test_lex_ordering_to_sorting_columns_uses_writer_schema() -> Result<()> { diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs index 3a1766869d319..52b0468b76a30 100644 --- a/datafusion/datasource-parquet/src/opener/mod.rs +++ b/datafusion/datasource-parquet/src/opener/mod.rs @@ -2163,6 +2163,7 @@ mod test { use datafusion_datasource::file_scan_config::FileScanConfigBuilder; use datafusion_datasource::morsel::{Morsel, Morselizer}; use datafusion_datasource::{PartitionedFile, TableSchema, TableSchemaBuilder}; + use datafusion_execution::cache::StoreScopedPath; use datafusion_execution::cache::cache_manager::{ CachedFileMetadataEntry, FileMetadataCache, }; @@ -4110,6 +4111,7 @@ mod test { .with_parquet_file_reader_factory(Arc::new( CachedParquetFileReaderFactory::new( Arc::clone(&store), + ObjectStoreUrl::local_filesystem(), Arc::clone(&cache), ), )) @@ -4162,7 +4164,10 @@ mod test { .await; let cache: Arc = Arc::new(DefaultCache::new(1024 * 1024)); let original = DFParquetMetadata::new(store.as_ref(), &file.object_meta) - .with_file_metadata_cache(Some(Arc::clone(&cache))) + .with_file_metadata_cache(Some(( + Arc::clone(&cache), + (ObjectStoreUrl::local_filesystem()).into(), + ))) .with_page_index_policy(Some(PageIndexPolicy::Optional)) .fetch_metadata() .await @@ -4178,6 +4183,7 @@ mod test { .with_parquet_file_reader_factory(Arc::new( CachedParquetFileReaderFactory::new( Arc::clone(&store), + ObjectStoreUrl::local_filesystem(), Arc::clone(&cache), ), )) @@ -4207,7 +4213,12 @@ mod test { prepared.loaded.reader_metadata.metadata() )); } - let cached = cache.get(&file.object_meta.location).unwrap(); + let cached = cache + .get(&StoreScopedPath::new( + ObjectStoreUrl::local_filesystem(), + file.object_meta.location.clone(), + )) + .unwrap(); let cached = cached .file_metadata .as_any() @@ -4305,6 +4316,7 @@ mod test { .with_parquet_file_reader_factory(Arc::new( CachedParquetFileReaderFactory::new( Arc::clone(&store), + ObjectStoreUrl::local_filesystem(), Arc::clone(&cache), ), )) @@ -4334,7 +4346,12 @@ mod test { }; assert_eq!(pruning_metrics.pruned(), 2); assert!( - cache.get(&file.object_meta.location).is_none(), + cache + .get(&StoreScopedPath::new( + ObjectStoreUrl::local_filesystem(), + file.object_meta.location.clone(), + )) + .is_none(), "encrypted metadata must not enter the shared cache" ); } @@ -5300,10 +5317,12 @@ mod test { #[tokio::test] async fn test_page_index_skipped_with_cached_reader_factory() { let store = Arc::new(InMemory::new()) as Arc; - let metadata_cache: Arc = - Arc::new(DefaultCache::::new( - 64 * 1024 * 1024, - )); + let metadata_cache: Arc = Arc::new(DefaultCache::< + StoreScopedPath, + CachedFileMetadataEntry, + >::new( + 64 * 1024 * 1024 + )); let values: Vec = (1..=100).collect(); let batch = record_batch!(( "a", @@ -5341,6 +5360,7 @@ mod test { .with_parquet_file_reader_factory(Arc::new( CachedParquetFileReaderFactory::new( Arc::clone(&store), + ObjectStoreUrl::local_filesystem(), Arc::clone(&metadata_cache), ), )) @@ -5352,7 +5372,10 @@ mod test { assert_eq!(counter_metric_value(&metrics, "page_index_load_skipped"), 1); let cached = metadata_cache - .get(&Path::from("test.parquet")) + .get(&StoreScopedPath::new( + ObjectStoreUrl::local_filesystem(), + Path::from("test.parquet"), + )) .expect("metadata cache should contain the file"); let extra_info = cached.file_metadata.extra_info(); let page_index_cached = extra_info.get("page_index").map(String::as_str); diff --git a/datafusion/datasource-parquet/src/reader.rs b/datafusion/datasource-parquet/src/reader.rs index 06ab40fdd27cd..1f0eb8800f3c4 100644 --- a/datafusion/datasource-parquet/src/reader.rs +++ b/datafusion/datasource-parquet/src/reader.rs @@ -26,6 +26,7 @@ use datafusion_common::HashMap; use datafusion_datasource::PartitionedFile; use datafusion_execution::cache::cache_manager::FileMetadata; use datafusion_execution::cache::cache_manager::FileMetadataCache; +use datafusion_execution::object_store::ObjectStoreUrl; use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet; use futures::FutureExt; use futures::TryFutureExt; @@ -143,16 +144,20 @@ impl ParquetFileReaderFactory for DefaultParquetFileReaderFactory { #[derive(Debug)] pub struct CachedParquetFileReaderFactory { store: Arc, + object_store_url: Arc, metadata_cache: Arc, } impl CachedParquetFileReaderFactory { + /// Create a factory whose metadata cache entries are scoped to `object_store_url`. pub fn new( store: Arc, + object_store_url: ObjectStoreUrl, metadata_cache: Arc, ) -> Self { Self { store, + object_store_url: Arc::new(object_store_url), metadata_cache, } } @@ -178,7 +183,10 @@ impl ParquetFileReaderFactory for CachedParquetFileReaderFactory { partitioned_file, ) .with_metadata_hint(metadata_size_hint) - .with_metadata_cache(Some(Arc::clone(&self.metadata_cache))); + .with_metadata_cache(Some(( + Arc::clone(&self.metadata_cache), + Arc::clone(&self.object_store_url), + ))); Ok(Box::new(reader)) } @@ -202,7 +210,7 @@ pub struct ParquetFileReader { file_metrics: ParquetFileMetrics, store: Arc, partitioned_file: PartitionedFile, - metadata_cache: Option>, + metadata_cache: Option<(Arc, Arc)>, metadata_size_hint: Option, } @@ -239,10 +247,10 @@ impl ParquetFileReader { &self.partitioned_file } - /// Set the [`FileMetadataCache`] for this reader + /// Set the [`FileMetadataCache`] and the URL identifying this reader's store. pub fn with_metadata_cache( mut self, - metadata_cache: Option>, + metadata_cache: Option<(Arc, Arc)>, ) -> Self { self.metadata_cache = metadata_cache; self @@ -306,9 +314,13 @@ impl AsyncFileReader for ParquetFileReader { let page_index_policy = options.map(|o| o.column_index_policy()); - DFParquetMetadata::new(&self.store, &object_meta) - .with_decryption_properties(file_decryption_properties) - .with_file_metadata_cache(metadata_cache) + let mut metadata = DFParquetMetadata::new(&self.store, &object_meta) + .with_decryption_properties(file_decryption_properties); + if let Some((cache, object_store_url)) = metadata_cache { + metadata = + metadata.with_file_metadata_cache(Some((cache, object_store_url))); + } + metadata .with_metadata_size_hint(self.metadata_size_hint) .with_page_index_policy(page_index_policy) .fetch_metadata() diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs index fbb6265847afc..eb913fc92e100 100644 --- a/datafusion/datasource-parquet/src/source.rs +++ b/datafusion/datasource-parquet/src/source.rs @@ -1253,14 +1253,17 @@ impl ParquetSource { let store = ctx .task_ctx() .runtime_env() - .object_store(object_store_url)?; + .object_store(&object_store_url)?; let metadata_cache = ctx .task_ctx() .runtime_env() .cache_manager .get_file_metadata_cache(); - let reader_factory = - Arc::new(CachedParquetFileReaderFactory::new(store, metadata_cache)); + let reader_factory = Arc::new(CachedParquetFileReaderFactory::new( + store, + object_store_url.clone(), + metadata_cache, + )); let mut source = ParquetSource::new(table_schema) .with_parquet_file_reader_factory(reader_factory) diff --git a/datafusion/datasource/src/file_format.rs b/datafusion/datasource/src/file_format.rs index 2c2a94e8cc4b4..b3c7d99edd30d 100644 --- a/datafusion/datasource/src/file_format.rs +++ b/datafusion/datasource/src/file_format.rs @@ -33,6 +33,8 @@ use datafusion_common::file_options::file_type::FileType; use datafusion_common::{ GetExt, Result, SchemaError, Statistics, internal_err, not_impl_err, schema_err, }; +use datafusion_execution::cache::cache_manager::FileMetadataCache; +use datafusion_execution::object_store::ObjectStoreUrl; use datafusion_physical_expr::LexRequirement; use datafusion_physical_expr_common::sort_expr::LexOrdering; use datafusion_physical_plan::ExecutionPlan; @@ -113,6 +115,18 @@ pub trait FileFormat: Any + Send + Sync + fmt::Debug { /// Returns whether this instance uses compression if applicable fn compression_type(&self) -> Option; + /// Return a format configured with a store-scoped metadata cache. + /// + /// The default returns `None`, so formats that do not use this cache keep + /// their existing inference implementation. Callers then use the original + /// format. Keeping the cache and its URL together prevents unscoped reuse. + fn with_file_metadata_cache( + &self, + _cache: Option<(Arc, Arc)>, + ) -> Option> { + None + } + /// Infer the common schema of the provided objects. The objects will usually /// be analysed up to a given number of records or files (as specified in the /// format config) then give the estimated common schema. This might fail if diff --git a/datafusion/datasource/src/url.rs b/datafusion/datasource/src/url.rs index 50d277803e38e..cec5f348b4cdf 100644 --- a/datafusion/datasource/src/url.rs +++ b/datafusion/datasource/src/url.rs @@ -19,7 +19,7 @@ use std::sync::Arc; use datafusion_common::{DataFusionError, Result, TableReference}; use datafusion_execution::cache::cache_manager::CachedFileList; -use datafusion_execution::cache::cache_manager::TableScopedPath; +use datafusion_execution::cache::cache_manager::{StoreScopedPath, TableScopedPath}; use datafusion_execution::object_store::ObjectStoreUrl; use datafusion_session::Session; @@ -267,6 +267,7 @@ impl ListingTableUrl { list_with_cache( ctx, store, + self.object_store(), self.table_ref.as_ref(), &self.prefix, prefix.as_ref(), @@ -283,6 +284,7 @@ impl ListingTableUrl { list_with_cache( ctx, store, + self.object_store(), self.table_ref.as_ref(), &self.prefix, prefix.as_ref(), @@ -364,6 +366,8 @@ impl ListingTableUrl { /// # Arguments /// * `ctx` - The session context /// * `store` - The object store to list from +/// * `object_store_url` - The registered store URL used to scope cache entries +/// * `table_ref` - Optional table reference used for table-level cache invalidation /// * `table_base_path` - The table's base path (the stable cache key) /// * `prefix` - Optional prefix relative to table base for filtering results /// @@ -373,6 +377,7 @@ impl ListingTableUrl { async fn list_with_cache<'b>( ctx: &'b dyn Session, store: &'b dyn ObjectStore, + object_store_url: ObjectStoreUrl, table_ref: Option<&TableReference>, table_base_path: &Path, prefix: Option<&Path>, @@ -393,15 +398,22 @@ async fn list_with_cache<'b>( .map(|res| res.map_err(|e| DataFusionError::ObjectStore(Box::new(e)))) .boxed()), Some(cache) => { + let object_store_url = Arc::new(object_store_url); let filter_prefix = prefix.is_some().then(|| full_prefix.clone()); let table_scoped_base_path = TableScopedPath { table: table_ref.cloned(), - path: table_base_path.clone(), + store_path: StoreScopedPath::new( + Arc::clone(&object_store_url), + table_base_path.clone(), + ), }; let table_scoped_list_path = TableScopedPath { table: table_ref.cloned(), - path: full_prefix.clone(), + store_path: StoreScopedPath::new( + Arc::clone(&object_store_url), + full_prefix.clone(), + ), }; let vec = if let Some(cached) = cache.get(&table_scoped_base_path) { @@ -854,7 +866,10 @@ mod tests { create_file(&store, "/table/year=2024/month=12/data5.parquet").await; // Session WITHOUT cache - let session_no_cache = MockSession::new(); + let runtime_no_cache = RuntimeEnvBuilder::new() + .with_object_list_cache_limit(0) + .build_arc()?; + let session_no_cache = MockSession::with_runtime_env(runtime_no_cache); // Session WITH cache - use RuntimeEnvBuilder with cache limit (no TTL needed for this test) let runtime_with_cache = RuntimeEnvBuilder::new() @@ -862,33 +877,43 @@ mod tests { .build_arc()?; let session_with_cache = MockSession::with_runtime_env(runtime_with_cache); - // Test cases: (url, prefix, description) + // Test cases: (url, prefix, description, expected file count) let test_cases = vec![ - ("/table/", None, "full table listing"), + ("mem://bucket/table/", None, "full table listing", 5), ( - "/table/", + "mem://bucket/table/", Some(Path::from("year=2023")), "single partition filter", + 2, ), ( - "/table/", + "mem://bucket/table/", Some(Path::from("year=2024")), "different partition filter", + 3, ), ( - "/table/", + "mem://bucket/table/", Some(Path::from("year=2024/month=06")), "nested partition filter", + 1, ), ( - "/table/", + "mem://bucket/table/", Some(Path::from("year=2025")), "non-existent partition", + 0, ), ]; - for (url_str, prefix, description) in test_cases { + for (url_str, prefix, description, expected_count) in test_cases { let url = ListingTableUrl::parse(url_str)?; + session_with_cache + .runtime_env + .cache_manager + .get_list_files_cache() + .unwrap() + .clear(); // Get results WITHOUT cache (sorted for comparison) let mut results_no_cache: Vec = url @@ -900,6 +925,7 @@ mod tests { .map(|m| m.location.to_string()) .collect(); results_no_cache.sort(); + assert_eq!(results_no_cache.len(), expected_count, "{description}"); // Get results WITH cache (first call - cache miss, sorted for comparison) let mut results_with_cache_miss: Vec = url @@ -942,6 +968,40 @@ mod tests { Ok(()) } + #[tokio::test] + async fn test_listing_cache_separates_object_stores() -> Result<()> { + use datafusion_execution::runtime_env::RuntimeEnvBuilder; + + let first = MockObjectStore::new(); + let second = MockObjectStore::new(); + create_file(&first, "table/first.parquet").await; + create_file(&second, "table/second.parquet").await; + let runtime = RuntimeEnvBuilder::new() + .with_object_list_cache_limit(1024 * 1024) + .build_arc()?; + let session = MockSession::with_runtime_env(runtime); + + // Reuse both listings to verify that each store has its own cache entry. + for _ in 0..2 { + for (bucket, store, expected) in [ + ("first", &first, "table/first.parquet"), + ("second", &second, "table/second.parquet"), + ] { + let url = ListingTableUrl::parse(format!("s3://{bucket}/table/"))?; + let files = url + .list_prefixed_files(&session, store, None, "parquet") + .await? + .try_collect::>() + .await?; + assert_eq!(files.len(), 1); + assert_eq!(files[0].location.as_ref(), expected); + } + } + assert_eq!(first.list_prefixes(), vec![Some(Path::from("table"))]); + assert_eq!(second.list_prefixes(), vec![Some(Path::from("table"))]); + Ok(()) + } + #[tokio::test] async fn test_cache_miss_with_prefix_lists_prefixed_path() -> Result<()> { use datafusion_execution::runtime_env::RuntimeEnvBuilder; @@ -955,7 +1015,7 @@ mod tests { .with_object_list_cache_limit(1024 * 1024) .build_arc()?; let session = MockSession::with_runtime_env(runtime); - let url = ListingTableUrl::parse("/table/")?; + let url = ListingTableUrl::parse("mem://bucket/table/")?; let prefix = Path::from("year=2024/month=06"); let results: Vec = url @@ -1009,7 +1069,7 @@ mod tests { .options_mut() .execution .listing_table_ignore_subdirectory = false; - let url = ListingTableUrl::parse("/sales/")?; + let url = ListingTableUrl::parse("mem://bucket/sales/")?; let q1_results: Vec = url .list_prefixed_files( @@ -1086,7 +1146,7 @@ mod tests { .build_arc()?; let session = MockSession::with_runtime_env(runtime); - let url = ListingTableUrl::parse("/sales/")?; + let url = ListingTableUrl::parse("mem://bucket/sales/")?; // First: query full table (populates cache) let full_results: Vec = url diff --git a/datafusion/execution/src/cache/cache_manager.rs b/datafusion/execution/src/cache/cache_manager.rs index bbbcaa4c77559..b47683ea37ecf 100644 --- a/datafusion/execution/src/cache/cache_manager.rs +++ b/datafusion/execution/src/cache/cache_manager.rs @@ -16,7 +16,9 @@ // under the License. use crate::cache::default_cache::DefaultCache; -pub use crate::cache::{Cache, CacheValue, SchemaFingerprint, TableScopedPath}; +pub use crate::cache::{ + Cache, CacheValue, SchemaFingerprint, StoreScopedPath, TableScopedPath, +}; use datafusion_common::HashMap; use datafusion_common::heap_size::{DFHeapSize, DFHeapSizeCtx}; use datafusion_common::{Result, Statistics}; @@ -28,6 +30,7 @@ use std::fmt::{Debug, Formatter}; use std::ops::Deref; use std::sync::Arc; use std::time::Duration; +use url::Url; pub const DEFAULT_LIST_FILES_CACHE_MEMORY_LIMIT: usize = 1024 * 1024; // 1MiB @@ -40,7 +43,7 @@ pub const DEFAULT_METADATA_CACHE_LIMIT: usize = 50 * 1024 * 1024; // 50M /// A cache for file statistics and orderings. /// /// This cache stores [`CachedFileMetadata`] which includes: -/// - File metadata for validation (size, last_modified) +/// - File metadata for validation (size, last_modified, e_tag, version) /// - Statistics for the file /// - Ordering information for the file /// @@ -75,6 +78,7 @@ pub type ListFilesCache = dyn Cache; /// /// This cache stores per-file metadata in the form of [`CachedFileMetadataEntry`], /// which includes the [`ObjectMeta`] for validation. +/// Entries are keyed by the object store URL and the path within that store. /// /// For example, the built in [`ListingTable`] uses this cache to avoid parsing /// Parquet footers multiple times for the same file. @@ -87,7 +91,7 @@ pub type ListFilesCache = dyn Cache; /// See [`crate::runtime_env::RuntimeEnv`] for more details. /// /// [`ListingTable`]: https://docs.rs/datafusion/latest/datafusion/datasource/listing/struct.ListingTable.html -pub type FileMetadataCache = dyn Cache; +pub type FileMetadataCache = dyn Cache; /// Cached metadata for a file, including statistics and ordering. /// @@ -95,7 +99,7 @@ pub type FileMetadataCache = dyn Cache; /// 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). + /// File metadata used for cache validation (size, last_modified, e_tag, version). pub meta: ObjectMeta, /// Fingerprint of the `file_schema` used to compute `statistics`. pub schema_fingerprint: Arc, @@ -123,20 +127,45 @@ impl CachedFileMetadata { /// Check if this cached entry is still valid for the given metadata. /// - /// Returns true if the file size, last modified time, and schema match. + /// Size and last modified time must match, along with a shared strong ETag + /// or, when unavailable, a shared version. The schema must always match. pub fn is_valid_for( &self, current_meta: &ObjectMeta, current_schema_fingerprint: &Arc, ) -> bool { - self.meta.size == current_meta.size - && self.meta.last_modified == current_meta.last_modified + file_metadata_matches(&self.meta, current_meta) && (Arc::ptr_eq(&self.schema_fingerprint, current_schema_fingerprint) || self.schema_fingerprint.as_ref() == current_schema_fingerprint.as_ref()) } } +/// Require size and time to match, then compare content or version identity +/// when available. +fn file_metadata_matches(cached: &ObjectMeta, current: &ObjectMeta) -> bool { + if cached.size != current.size || cached.last_modified != current.last_modified { + return false; + } + // The HTTP object store passes through weak ETags, which do not guarantee + // identical bytes. Treat them as missing and try the remaining validators. + if let (Some(cached), Some(current)) = + (cached.e_tag.as_deref(), current.e_tag.as_deref()) + && !cached.starts_with("W/") + && !current.starts_with("W/") + { + // Compare opaque values verbatim. Some stores (e.g. Azure) return + // unquoted ETags in listings and quoted ETags in HEAD responses. This + // can cause conservative cache misses, but cannot reuse stale metadata. + return cached == current; + } + if let Some((cached, current)) = cached.version.as_ref().zip(current.version.as_ref()) + { + return cached == current; + } + true +} + impl CacheValue for CachedFileMetadata { fn size(&self) -> usize { DFHeapSize::heap_size(self, &mut DFHeapSizeCtx::default()) @@ -259,7 +288,7 @@ pub trait FileMetadata: Any + Send + Sync { /// Cached file metadata entry with validation information. #[derive(Clone)] pub struct CachedFileMetadataEntry { - /// File metadata used for cache validation (size, last_modified). + /// File metadata used for cache validation (size, last_modified, e_tag, version). pub meta: ObjectMeta, /// The cached file metadata. pub file_metadata: Arc, @@ -281,9 +310,11 @@ impl CachedFileMetadataEntry { } /// Check if this cached entry is still valid for the given metadata. + /// + /// Requires matching size and last modified time, plus a shared strong ETag + /// or, when unavailable, a shared version. Weak ETags are treated as missing. pub fn is_valid_for(&self, current_meta: &ObjectMeta) -> bool { - self.meta.size == current_meta.size - && self.meta.last_modified == current_meta.last_modified + file_metadata_matches(&self.meta, current_meta) } } @@ -311,6 +342,36 @@ pub struct CacheManager { } impl CacheManager { + /// Invalidate entries for the registry key of `url`, including anonymous + /// and named tables. Credentials and paths do not affect registration. + pub(crate) fn invalidate_object_store(&self, url: &Url) { + let matches = |store: &crate::object_store::ObjectStoreUrl| { + let store: &Url = store.as_ref(); + store.scheme() == url.scheme() + && store.host() == url.host() + && store.port() == url.port() + }; + for key in self.file_metadata_cache.list_entries().keys() { + if matches(&key.object_store_url) { + self.file_metadata_cache.remove(key); + } + } + if let Some(cache) = &self.file_statistic_cache { + for key in cache.list_entries().keys() { + if matches(&key.store_path.object_store_url) { + cache.remove(key); + } + } + } + if let Some(cache) = &self.list_files_cache { + for key in cache.list_entries().keys() { + if matches(&key.store_path.object_store_url) { + cache.remove(key); + } + } + } + } + pub fn try_new(config: &CacheManagerConfig) -> Result> { let file_statistic_cache: Option> = match &config.file_statistics_cache { @@ -512,6 +573,185 @@ impl CacheManagerConfig { #[cfg(test)] mod tests { use super::*; + use arrow::datatypes::Schema; + + struct TestFileMetadata; + + impl FileMetadata for TestFileMetadata { + fn as_any(&self) -> &dyn Any { + self + } + + fn memory_size(&self) -> usize { + 0 + } + + fn extra_info(&self) -> HashMap { + HashMap::new() + } + } + + fn test_object_meta() -> ObjectMeta { + ObjectMeta { + location: Path::from("data.parquet"), + last_modified: chrono::DateTime::UNIX_EPOCH, + size: 100, + e_tag: None, + version: None, + } + } + + fn assert_metadata_validity( + cached_meta: &ObjectMeta, + current_meta: &ObjectMeta, + expected: bool, + ) { + let schema = Schema::empty(); + let fingerprint = Arc::new(SchemaFingerprint::from_schema(&schema)); + let cached_statistics = CachedFileMetadata::new( + cached_meta.clone(), + Arc::clone(&fingerprint), + Arc::new(Statistics::new_unknown(&schema)), + None, + ); + let cached_metadata = + CachedFileMetadataEntry::new(cached_meta.clone(), Arc::new(TestFileMetadata)); + + assert_eq!( + cached_statistics.is_valid_for(current_meta, &fingerprint), + expected, + "statistics: cached={cached_meta:?}, current={current_meta:?}" + ); + assert_eq!( + cached_metadata.is_valid_for(current_meta), + expected, + "metadata: cached={cached_meta:?}, current={current_meta:?}" + ); + } + + #[test] + fn test_metadata_validation_etag_and_version() { + let etags = [ + (Some("same"), Some("same"), Some(true)), + (Some("old"), Some("new"), Some(false)), + (Some("old"), None, None), + (None, Some("new"), None), + (None, None, None), + (Some("W/\"same\""), Some("W/\"same\""), None), + (Some("W/\"old\""), Some("W/\"new\""), None), + (Some("W/\"same\""), Some("\"same\""), None), + (Some("\"same\""), Some("W/\"same\""), None), + // Azure listing and HEAD values remain opaque, not normalized. + ( + Some("0x8D93C7D4629C227"), + Some("\"0x8D93C7D4629C227\""), + Some(false), + ), + ]; + let versions = [ + (Some("same"), Some("same"), Some(true)), + (Some("old"), Some("new"), Some(false)), + (Some("old"), None, None), + (None, Some("new"), None), + (None, None, None), + ]; + + for (cached_etag, current_etag, etag_matches) in etags { + for (cached_version, current_version, version_matches) in versions { + let cached_meta = ObjectMeta { + e_tag: cached_etag.map(str::to_owned), + version: cached_version.map(str::to_owned), + ..test_object_meta() + }; + let current_meta = ObjectMeta { + e_tag: current_etag.map(str::to_owned), + version: current_version.map(str::to_owned), + ..test_object_meta() + }; + assert_metadata_validity( + &cached_meta, + ¤t_meta, + etag_matches.or(version_matches).unwrap_or(true), + ); + } + } + } + + #[test] + fn test_metadata_validation_size_and_last_modified() { + // Missing, weak, and one-sided identifiers all use the fallback. + for (cached_etag, current_etag) in [ + (None, None), + (Some("W/\"same\""), Some("W/\"same\"")), + (Some("same"), None), + (None, Some("same")), + (Some("W/\"same\""), Some("\"same\"")), + (Some("\"same\""), Some("W/\"same\"")), + ] { + let cached_meta = ObjectMeta { + e_tag: cached_etag.map(str::to_owned), + ..test_object_meta() + }; + let current_meta = ObjectMeta { + e_tag: current_etag.map(str::to_owned), + ..test_object_meta() + }; + assert_metadata_validity(&cached_meta, ¤t_meta, true); + let mut current_meta = current_meta; + current_meta.size += 1; + assert_metadata_validity(&cached_meta, ¤t_meta, false); + + current_meta.size = cached_meta.size; + current_meta.last_modified += chrono::Duration::seconds(1); + assert_metadata_validity(&cached_meta, ¤t_meta, false); + } + } + + #[test] + fn test_metadata_validation_identifiers_also_require_size_and_time() { + for (etag, cached_version, current_version) in [ + (Some("same"), Some("old"), Some("new")), + (None, Some("same"), Some("same")), + (Some("W/\"same\""), Some("same"), Some("same")), + ] { + let cached_meta = ObjectMeta { + e_tag: etag.map(str::to_owned), + version: cached_version.map(str::to_owned), + ..test_object_meta() + }; + let mut current_meta = ObjectMeta { + size: cached_meta.size + 1, + version: current_version.map(str::to_owned), + ..cached_meta.clone() + }; + assert_metadata_validity(&cached_meta, ¤t_meta, false); + current_meta.size = cached_meta.size; + current_meta.last_modified += chrono::Duration::seconds(1); + assert_metadata_validity(&cached_meta, ¤t_meta, false); + } + } + + #[test] + fn test_matching_etag_does_not_override_schema_validation() { + use arrow::datatypes::{DataType, Field}; + + let meta = ObjectMeta { + e_tag: Some("same".into()), + ..test_object_meta() + }; + let original = Schema::new(vec![Field::new("value", DataType::Int32, false)]); + let changed = Schema::new(vec![Field::new("value", DataType::Int64, false)]); + let cached = CachedFileMetadata::new( + meta.clone(), + Arc::new(SchemaFingerprint::from_schema(&original)), + Arc::new(Statistics::new_unknown(&original)), + None, + ); + assert!( + !cached + .is_valid_for(&meta, &Arc::new(SchemaFingerprint::from_schema(&changed))) + ); + } /// Test to verify that TTL is preserved when not explicitly set in config. /// This fixes issue #19396 where TTL was being unset from DefaultListFilesCache diff --git a/datafusion/execution/src/cache/default_cache.rs b/datafusion/execution/src/cache/default_cache.rs index bfe326f3a47e1..25799ebe1782b 100644 --- a/datafusion/execution/src/cache/default_cache.rs +++ b/datafusion/execution/src/cache/default_cache.rs @@ -310,7 +310,8 @@ mod tests { use crate::cache::default_cache::TimeProvider; use crate::cache::{Cache, CacheEntryInfo}; use crate::cache::{CacheKey, CacheValue}; - use crate::cache::{SchemaFingerprint, TableScopedPath}; + use crate::cache::{SchemaFingerprint, StoreScopedPath, TableScopedPath}; + use crate::object_store::ObjectStoreUrl; use arrow::array::{Int32Array, ListArray, RecordBatch}; use arrow::buffer::{OffsetBuffer, ScalarBuffer}; use arrow::datatypes::{DataType, Field, Schema, TimeUnit}; @@ -366,6 +367,56 @@ mod tests { } } + fn metadata_cache_key(path: &Path) -> StoreScopedPath { + StoreScopedPath::new(ObjectStoreUrl::local_filesystem(), path.clone()) + } + + #[test] + fn test_file_metadata_cache_separates_object_stores() { + let cache = DefaultCache::new(1024); + let meta = create_test_object_meta("data.parquet", 100); + let first_key = StoreScopedPath::new( + ObjectStoreUrl::parse("s3://first").unwrap(), + meta.location.clone(), + ); + let second_key = StoreScopedPath::new( + ObjectStoreUrl::parse("s3://second").unwrap(), + meta.location.clone(), + ); + let first: Arc = Arc::new(TestFileMetadata { + metadata: "first".to_owned(), + }); + let second: Arc = Arc::new(TestFileMetadata { + metadata: "second".to_owned(), + }); + + cache.put( + &first_key, + CachedFileMetadataEntry::new(meta.clone(), Arc::clone(&first)), + ); + assert!(cache.get(&second_key).is_none()); + cache.put( + &second_key, + CachedFileMetadataEntry::new(meta, Arc::clone(&second)), + ); + assert_eq!(cache.len(), 2); + assert!(Arc::ptr_eq( + &cache.get(&first_key).unwrap().file_metadata, + &first + )); + assert!(Arc::ptr_eq( + &cache.get(&second_key).unwrap().file_metadata, + &second + )); + assert_eq!( + cache.memory_used(), + first_key.size() + + second_key.size() + + first.memory_size() + + second.memory_size() + ); + } + #[test] fn test_default_file_metadata_cache() { let object_meta = create_test_object_meta("test", 1024); @@ -377,50 +428,62 @@ mod tests { let cache = DefaultCache::new(1024 * 1024); // Cache miss - assert!(cache.get(&object_meta.location).is_none()); + assert!( + cache + .get(&metadata_cache_key(&object_meta.location)) + .is_none() + ); // Put a value let cached_entry = CachedFileMetadataEntry::new(object_meta.clone(), Arc::clone(&metadata)); - cache.put(&object_meta.location, cached_entry); + cache.put(&metadata_cache_key(&object_meta.location), cached_entry); // Verify the cached value - assert!(cache.contains_key(&object_meta.location)); - let result = cache.get(&object_meta.location).unwrap(); + assert!(cache.contains_key(&metadata_cache_key(&object_meta.location))); + let result = cache + .get(&metadata_cache_key(&object_meta.location)) + .unwrap(); let test_file_metadata = Arc::downcast::(result.file_metadata); assert!(test_file_metadata.is_ok()); assert_eq!(test_file_metadata.unwrap().metadata, "retrieved_metadata"); // Cache hit - check validation - let result2 = cache.get(&object_meta.location).unwrap(); + let result2 = cache + .get(&metadata_cache_key(&object_meta.location)) + .unwrap(); assert!(result2.is_valid_for(&object_meta)); // File size changed - closure should detect invalidity let object_meta2 = create_test_object_meta("test", 2048); - let result3 = cache.get(&object_meta2.location).unwrap(); + let result3 = cache + .get(&metadata_cache_key(&object_meta2.location)) + .unwrap(); // Cached entry should NOT be valid for new meta assert!(!result3.is_valid_for(&object_meta2)); // Return new entry let new_entry = CachedFileMetadataEntry::new(object_meta2.clone(), Arc::clone(&metadata)); - cache.put(&object_meta2.location, new_entry); + cache.put(&metadata_cache_key(&object_meta2.location), new_entry); - let result4 = cache.get(&object_meta2.location).unwrap(); + let result4 = cache + .get(&metadata_cache_key(&object_meta2.location)) + .unwrap(); assert_eq!(result4.meta.size, 2048); // remove - cache.remove(&object_meta.location); - assert!(!cache.contains_key(&object_meta.location)); + cache.remove(&metadata_cache_key(&object_meta.location)); + assert!(!cache.contains_key(&metadata_cache_key(&object_meta.location))); // len and clear let object_meta3 = create_test_object_meta("test3", 100); cache.put( - &object_meta.location, + &metadata_cache_key(&object_meta.location), CachedFileMetadataEntry::new(object_meta.clone(), Arc::clone(&metadata)), ); cache.put( - &object_meta3.location, + &metadata_cache_key(&object_meta3.location), CachedFileMetadataEntry::new(object_meta3.clone(), Arc::clone(&metadata)), ); assert_eq!(cache.len(), 2); @@ -447,86 +510,87 @@ mod tests { #[test] fn test_default_file_metadata_cache_with_limit() { - // Create a cache with 1000 bytes capacity + 4 keys each key 2 bytes - let cache = DefaultCache::new(1000 + 4 * 2); + // Allow 1000 bytes of metadata plus four store-qualified keys. + let key_size = metadata_cache_key(&Path::from("01")).size(); + let cache = DefaultCache::new(1000 + 4 * key_size); let (object_meta1, metadata1) = generate_test_metadata_with_size("01", 100); let (object_meta2, metadata2) = generate_test_metadata_with_size("02", 500); let (object_meta3, metadata3) = generate_test_metadata_with_size("03", 300); cache.put( - &object_meta1.location, + &metadata_cache_key(&object_meta1.location), CachedFileMetadataEntry::new(object_meta1.clone(), metadata1), ); cache.put( - &object_meta2.location, + &metadata_cache_key(&object_meta2.location), CachedFileMetadataEntry::new(object_meta2.clone(), metadata2), ); cache.put( - &object_meta3.location, + &metadata_cache_key(&object_meta3.location), CachedFileMetadataEntry::new(object_meta3.clone(), metadata3), ); // all entries will fit assert_eq!(cache.len(), 3); - assert_eq!(cache.memory_used(), 906); - assert!(cache.contains_key(&object_meta1.location)); - assert!(cache.contains_key(&object_meta2.location)); - assert!(cache.contains_key(&object_meta3.location)); + assert_eq!(cache.memory_used(), 900 + 3 * key_size); + assert!(cache.contains_key(&metadata_cache_key(&object_meta1.location))); + assert!(cache.contains_key(&metadata_cache_key(&object_meta2.location))); + assert!(cache.contains_key(&metadata_cache_key(&object_meta3.location))); // add a new entry which will remove the least recently used ("1") let (object_meta4, metadata4) = generate_test_metadata_with_size("04", 200); cache.put( - &object_meta4.location, + &metadata_cache_key(&object_meta4.location), CachedFileMetadataEntry::new(object_meta4.clone(), metadata4), ); assert_eq!(cache.len(), 3); - assert_eq!(cache.memory_used(), 1006); - assert!(!cache.contains_key(&object_meta1.location)); - assert!(cache.contains_key(&object_meta4.location)); + assert_eq!(cache.memory_used(), 1000 + 3 * key_size); + assert!(!cache.contains_key(&metadata_cache_key(&object_meta1.location))); + assert!(cache.contains_key(&metadata_cache_key(&object_meta4.location))); // get entry "2", which will move it to the top of the queue, and add a new one which will // remove the new least recently used ("3") - let _ = cache.get(&object_meta2.location); + let _ = cache.get(&metadata_cache_key(&object_meta2.location)); let (object_meta5, metadata5) = generate_test_metadata_with_size("05", 100); cache.put( - &object_meta5.location, + &metadata_cache_key(&object_meta5.location), CachedFileMetadataEntry::new(object_meta5.clone(), metadata5), ); assert_eq!(cache.len(), 3); - assert_eq!(cache.memory_used(), 806); - assert!(!cache.contains_key(&object_meta3.location)); - assert!(cache.contains_key(&object_meta5.location)); + assert_eq!(cache.memory_used(), 800 + 3 * key_size); + assert!(!cache.contains_key(&metadata_cache_key(&object_meta3.location))); + assert!(cache.contains_key(&metadata_cache_key(&object_meta5.location))); // new entry which will not be able to fit in the 1000 bytes allocated let (object_meta6, metadata6) = generate_test_metadata_with_size("06", 1200); cache.put( - &object_meta6.location, + &metadata_cache_key(&object_meta6.location), CachedFileMetadataEntry::new(object_meta6.clone(), metadata6), ); assert_eq!(cache.len(), 3); - assert_eq!(cache.memory_used(), 806); - assert!(!cache.contains_key(&object_meta6.location)); + assert_eq!(cache.memory_used(), 800 + 3 * key_size); + assert!(!cache.contains_key(&metadata_cache_key(&object_meta6.location))); // new entry which is able to fit without removing any entry let (object_meta7, metadata7) = generate_test_metadata_with_size("07", 200); cache.put( - &object_meta7.location, + &metadata_cache_key(&object_meta7.location), CachedFileMetadataEntry::new(object_meta7.clone(), metadata7), ); assert_eq!(cache.len(), 4); - assert_eq!(cache.memory_used(), 1008); - assert!(cache.contains_key(&object_meta7.location)); + assert_eq!(cache.memory_used(), 1000 + 4 * key_size); + assert!(cache.contains_key(&metadata_cache_key(&object_meta7.location))); // new entry which will remove all other entries let (object_meta8, metadata8) = generate_test_metadata_with_size("08", 999); cache.put( - &object_meta8.location, + &metadata_cache_key(&object_meta8.location), CachedFileMetadataEntry::new(object_meta8.clone(), metadata8), ); assert_eq!(cache.len(), 1); - assert_eq!(cache.memory_used(), 1001); - assert!(cache.contains_key(&object_meta8.location)); + assert_eq!(cache.memory_used(), 999 + key_size); + assert!(cache.contains_key(&metadata_cache_key(&object_meta8.location))); // when updating an entry, the previous ones are not unnecessarily removed let (object_meta9, metadata9) = generate_test_metadata_with_size("09", 300); @@ -534,49 +598,49 @@ mod tests { let (object_meta11_v1, metadata11_v1) = generate_test_metadata_with_size("11", 400); cache.put( - &object_meta9.location, + &metadata_cache_key(&object_meta9.location), CachedFileMetadataEntry::new(object_meta9.clone(), metadata9), ); cache.put( - &object_meta10.location, + &metadata_cache_key(&object_meta10.location), CachedFileMetadataEntry::new(object_meta10.clone(), metadata10), ); cache.put( - &object_meta11_v1.location, + &metadata_cache_key(&object_meta11_v1.location), CachedFileMetadataEntry::new(object_meta11_v1.clone(), metadata11_v1), ); - assert_eq!(cache.memory_used(), 906); + assert_eq!(cache.memory_used(), 900 + 3 * key_size); assert_eq!(cache.len(), 3); let (object_meta11_v2, metadata11_v2) = generate_test_metadata_with_size("11", 500); cache.put( - &object_meta11_v2.location, + &metadata_cache_key(&object_meta11_v2.location), CachedFileMetadataEntry::new(object_meta11_v2.clone(), metadata11_v2), ); - assert_eq!(cache.memory_used(), 1006); + assert_eq!(cache.memory_used(), 1000 + 3 * key_size); assert_eq!(cache.len(), 3); - assert!(cache.contains_key(&object_meta9.location)); - assert!(cache.contains_key(&object_meta10.location)); - assert!(cache.contains_key(&object_meta11_v2.location)); + assert!(cache.contains_key(&metadata_cache_key(&object_meta9.location))); + assert!(cache.contains_key(&metadata_cache_key(&object_meta10.location))); + assert!(cache.contains_key(&metadata_cache_key(&object_meta11_v2.location))); // when updating an entry that now exceeds the limit, the LRU ("09") needs to be removed let (object_meta11_v3, metadata11_v3) = - generate_test_metadata_with_size("11", 510); + generate_test_metadata_with_size("11", 510 + key_size); cache.put( - &object_meta11_v3.location, + &metadata_cache_key(&object_meta11_v3.location), CachedFileMetadataEntry::new(object_meta11_v3.clone(), metadata11_v3), ); - assert_eq!(cache.memory_used(), 714); + assert_eq!(cache.memory_used(), 710 + 3 * key_size); assert_eq!(cache.len(), 2); - assert!(cache.contains_key(&object_meta10.location)); - assert!(cache.contains_key(&object_meta11_v3.location)); + assert!(cache.contains_key(&metadata_cache_key(&object_meta10.location))); + assert!(cache.contains_key(&metadata_cache_key(&object_meta11_v3.location))); // manually removing an entry that is not the LRU - cache.remove(&object_meta11_v3.location); + cache.remove(&metadata_cache_key(&object_meta11_v3.location)); assert_eq!(cache.len(), 1); - assert_eq!(cache.memory_used(), 202); - assert!(cache.contains_key(&object_meta10.location)); - assert!(!cache.contains_key(&object_meta11_v3.location)); + assert_eq!(cache.memory_used(), 200 + key_size); + assert!(cache.contains_key(&metadata_cache_key(&object_meta10.location))); + assert!(!cache.contains_key(&metadata_cache_key(&object_meta11_v3.location))); // clear cache.clear(); @@ -588,31 +652,32 @@ mod tests { let (object_meta13, metadata13) = generate_test_metadata_with_size("13", 200); let (object_meta14, metadata14) = generate_test_metadata_with_size("14", 500); cache.put( - &object_meta12.location, + &metadata_cache_key(&object_meta12.location), CachedFileMetadataEntry::new(object_meta12.clone(), metadata12), ); cache.put( - &object_meta13.location, + &metadata_cache_key(&object_meta13.location), CachedFileMetadataEntry::new(object_meta13.clone(), metadata13), ); cache.put( - &object_meta14.location, + &metadata_cache_key(&object_meta14.location), CachedFileMetadataEntry::new(object_meta14.clone(), metadata14), ); assert_eq!(cache.len(), 3); - assert_eq!(cache.memory_used(), 1006); + assert_eq!(cache.memory_used(), 1000 + 3 * key_size); cache.update_cache_limit(600); assert_eq!(cache.len(), 1); - assert_eq!(cache.memory_used(), 502); - assert!(!cache.contains_key(&object_meta12.location)); - assert!(!cache.contains_key(&object_meta13.location)); - assert!(cache.contains_key(&object_meta14.location)); + assert_eq!(cache.memory_used(), 500 + key_size); + assert!(!cache.contains_key(&metadata_cache_key(&object_meta12.location))); + assert!(!cache.contains_key(&metadata_cache_key(&object_meta13.location))); + assert!(cache.contains_key(&metadata_cache_key(&object_meta14.location))); } #[test] fn test_default_file_metadata_cache_entries_info() { - // Create a cache with 1000 bytes + 4 bytes for 4 keys each key 1 byte - let cache = DefaultCache::new(1000 + 4); + // Allow 1000 bytes of metadata plus four store-qualified keys. + let key_size = metadata_cache_key(&Path::from("1")).size(); + let cache = DefaultCache::new(1000 + 4 * key_size); let (object_meta1, metadata1) = generate_test_metadata_with_size("1", 100); let (object_meta2, metadata2) = generate_test_metadata_with_size("2", 200); @@ -625,16 +690,16 @@ mod tests { // Build a cache which fits exactly these 3 entries - cache.put(&object_meta1.location, entry_1.clone()); - cache.put(&object_meta2.location, entry_2.clone()); - cache.put(&object_meta3.location, entry_3.clone()); + cache.put(&metadata_cache_key(&object_meta1.location), entry_1.clone()); + cache.put(&metadata_cache_key(&object_meta2.location), entry_2.clone()); + cache.put(&metadata_cache_key(&object_meta3.location), entry_3.clone()); let entries = cache.list_entries(); assert_eq!( entries, HashMap::from([ ( - Path::from("1"), + metadata_cache_key(&Path::from("1")), CacheEntryInfo { value: entry_1.clone(), size_bytes: 100, @@ -643,7 +708,7 @@ mod tests { } ), ( - Path::from("2"), + metadata_cache_key(&Path::from("2")), CacheEntryInfo { value: entry_2.clone(), size_bytes: 200, @@ -652,7 +717,7 @@ mod tests { } ), ( - Path::from("3"), + metadata_cache_key(&Path::from("3")), CacheEntryInfo { value: entry_3.clone(), size_bytes: 300, @@ -664,12 +729,12 @@ mod tests { ); // new hit on "1" - let _ = cache.get(&object_meta1.location); + let _ = cache.get(&metadata_cache_key(&object_meta1.location)); assert_eq!( cache.list_entries(), HashMap::from([ ( - Path::from("1"), + metadata_cache_key(&Path::from("1")), CacheEntryInfo { value: entry_1.clone(), size_bytes: 100, @@ -678,7 +743,7 @@ mod tests { } ), ( - Path::from("2"), + metadata_cache_key(&Path::from("2")), CacheEntryInfo { value: entry_2.clone(), size_bytes: 200, @@ -687,7 +752,7 @@ mod tests { } ), ( - Path::from("3"), + metadata_cache_key(&Path::from("3")), CacheEntryInfo { value: entry_3.clone(), size_bytes: 300, @@ -701,12 +766,12 @@ mod tests { // new entry, will evict "2" let (object_meta4, metadata4) = generate_test_metadata_with_size("4", 600); let entry_4 = CachedFileMetadataEntry::new(object_meta4.clone(), metadata4); - cache.put(&object_meta4.location, entry_4.clone()); + cache.put(&metadata_cache_key(&object_meta4.location), entry_4.clone()); assert_eq!( cache.list_entries(), HashMap::from([ ( - Path::from("1"), + metadata_cache_key(&Path::from("1")), CacheEntryInfo { value: entry_1.clone(), size_bytes: 100, @@ -715,7 +780,7 @@ mod tests { } ), ( - Path::from("3"), + metadata_cache_key(&Path::from("3")), CacheEntryInfo { value: entry_3.clone(), size_bytes: 300, @@ -724,7 +789,7 @@ mod tests { } ), ( - Path::from("4"), + metadata_cache_key(&Path::from("4")), CacheEntryInfo { value: entry_4.clone(), size_bytes: 600, @@ -739,12 +804,15 @@ mod tests { let (object_meta1_new, metadata1_new) = generate_test_metadata_with_size("1", 50); let entry_1 = CachedFileMetadataEntry::new(object_meta1_new.clone(), metadata1_new); - cache.put(&object_meta1_new.location, entry_1.clone()); + cache.put( + &metadata_cache_key(&object_meta1_new.location), + entry_1.clone(), + ); assert_eq!( cache.list_entries(), HashMap::from([ ( - Path::from("1"), + metadata_cache_key(&Path::from("1")), CacheEntryInfo { value: entry_1.clone(), size_bytes: 50, @@ -753,7 +821,7 @@ mod tests { } ), ( - Path::from("3"), + metadata_cache_key(&Path::from("3")), CacheEntryInfo { value: entry_3.clone(), size_bytes: 300, @@ -762,7 +830,7 @@ mod tests { } ), ( - Path::from("4"), + metadata_cache_key(&Path::from("4")), CacheEntryInfo { value: entry_4.clone(), size_bytes: 600, @@ -774,12 +842,12 @@ mod tests { ); // remove entry "4" - cache.remove(&object_meta4.location); + cache.remove(&metadata_cache_key(&object_meta4.location)); assert_eq!( cache.list_entries(), HashMap::from([ ( - Path::from("1"), + metadata_cache_key(&Path::from("1")), CacheEntryInfo { value: entry_1.clone(), size_bytes: 50, @@ -788,7 +856,7 @@ mod tests { } ), ( - Path::from("3"), + metadata_cache_key(&Path::from("3")), CacheEntryInfo { value: entry_3.clone(), size_bytes: 300, @@ -828,8 +896,11 @@ mod tests { )]); let path = TableScopedPath { - path: meta.location.clone(), table: None, + store_path: StoreScopedPath::new( + ObjectStoreUrl::local_filesystem(), + meta.location.clone(), + ), }; let schema_fingerprint = Arc::new(SchemaFingerprint::from_schema(&schema)); @@ -889,8 +960,11 @@ mod tests { assert_eq!(entries.len(), 1); let path_3 = TableScopedPath { - path: Path::from("test"), table: None, + store_path: StoreScopedPath::new( + ObjectStoreUrl::local_filesystem(), + Path::from("test"), + ), }; let entry = entries.get(&path_3).unwrap(); @@ -964,8 +1038,11 @@ mod tests { ); let path = TableScopedPath { - path: meta.location.clone(), table: None, + store_path: StoreScopedPath::new( + ObjectStoreUrl::local_filesystem(), + meta.location.clone(), + ), }; cache.put(&path, cached_value); @@ -993,8 +1070,11 @@ mod tests { fn test_cache_invalidation_on_file_modification() { let cache = DefaultCache::new(DEFAULT_FILE_STATISTICS_MEMORY_LIMIT); let path = TableScopedPath { - path: Path::from("test.parquet"), table: None, + store_path: StoreScopedPath::new( + ObjectStoreUrl::local_filesystem(), + Path::from("test.parquet"), + ), }; let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]); let schema_fingerprint = Arc::new(SchemaFingerprint::from_schema(&schema)); @@ -1035,15 +1115,18 @@ mod tests { fn test_ordering_cache_invalidation_on_file_modification() { let cache = DefaultCache::new(DEFAULT_FILE_STATISTICS_MEMORY_LIMIT); let path = TableScopedPath { - path: Path::from("test.parquet"), table: None, + store_path: StoreScopedPath::new( + ObjectStoreUrl::local_filesystem(), + Path::from("test.parquet"), + ), }; 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 { - location: path.path.clone(), + location: path.store_path.path.clone(), last_modified: DateTime::parse_from_rfc3339("2022-09-27T22:36:00+02:00") .unwrap() .into(), @@ -1067,7 +1150,7 @@ mod tests { // File modified (size changed) let meta_v2 = ObjectMeta { - location: path.path.clone(), + location: path.store_path.path.clone(), last_modified: DateTime::parse_from_rfc3339("2022-09-28T10:00:00+02:00") .unwrap() .into(), @@ -1115,8 +1198,11 @@ mod tests { ); let path_1 = TableScopedPath { - path: meta1.location.clone(), table: None, + store_path: StoreScopedPath::new( + ObjectStoreUrl::local_filesystem(), + meta1.location.clone(), + ), }; cache.put(&path_1, cached_value_1.clone()); @@ -1129,8 +1215,11 @@ mod tests { ); let path_2 = TableScopedPath { - path: meta2.location.clone(), table: None, + store_path: StoreScopedPath::new( + ObjectStoreUrl::local_filesystem(), + meta2.location.clone(), + ), }; cache.put(&path_2, cached_value_2.clone()); @@ -1172,7 +1261,10 @@ mod tests { let mut ctx = DFHeapSizeCtx::default(); - let limit_for_2_entries = meta_1.location.as_ref().heap_size(&mut ctx) + let limit_for_2_entries = 2 * ObjectStoreUrl::local_filesystem() + .as_str() + .heap_size(&mut ctx) + + meta_1.location.as_ref().heap_size(&mut ctx) + value_1.heap_size(&mut ctx) + meta_2.location.as_ref().heap_size(&mut ctx) + value_2.heap_size(&mut ctx); @@ -1180,13 +1272,19 @@ mod tests { // create a cache with a limit which fits exactly 2 entries let cache = DefaultCache::new(limit_for_2_entries); let path_1 = TableScopedPath { - path: meta_1.location.clone(), table: None, + store_path: StoreScopedPath::new( + ObjectStoreUrl::local_filesystem(), + meta_1.location.clone(), + ), }; let path_2 = TableScopedPath { - path: meta_2.location.clone(), table: None, + store_path: StoreScopedPath::new( + ObjectStoreUrl::local_filesystem(), + meta_2.location.clone(), + ), }; cache.put(&path_1, value_1.clone()); @@ -1201,8 +1299,11 @@ mod tests { assert_eq!(result_2.unwrap(), value_2); let path_3 = TableScopedPath { - path: meta_3.location.clone(), table: None, + store_path: StoreScopedPath::new( + ObjectStoreUrl::local_filesystem(), + meta_3.location.clone(), + ), }; // adding the third entry evicts the first entry @@ -1230,7 +1331,7 @@ mod tests { assert_eq!(cache.len(), 1); assert_eq!( cache.memory_used(), - meta_3.location.as_ref().heap_size(&mut ctx) + value_3.heap_size(&mut ctx) + path_3.size() + value_3.heap_size(&mut ctx) ); cache.clear(); @@ -1249,8 +1350,11 @@ mod tests { let cache = DefaultCache::new(limit_less_than_the_entry); let path_1 = TableScopedPath { - path: meta.location.clone(), table: None, + store_path: StoreScopedPath::new( + ObjectStoreUrl::local_filesystem(), + meta.location.clone(), + ), }; cache.put(&path_1, value_too_large.clone()); @@ -1263,7 +1367,7 @@ mod tests { cache.put(&path_1, value_fits.clone()); assert_eq!(cache.len(), 1); - assert_eq!(cache.memory_used(), 1514); + assert_eq!(cache.memory_used(), path_1.size() + value_fits.size()); // now add an entry which is over the limit and make sure the old stale entry is removed let stale_entry = cache.put(&path_1, value_too_large.clone()); @@ -1364,7 +1468,10 @@ mod tests { ) -> (TableScopedPath, CachedFileList) { let key = TableScopedPath { table, - path: Path::from(path), + store_path: StoreScopedPath::new( + ObjectStoreUrl::local_filesystem(), + Path::from(path), + ), }; let metas: Vec = (0..count) .map(|i| create_object_meta(&format!("file{i}"), meta_size)) @@ -1380,7 +1487,7 @@ mod tests { let path = Path::from("test_path"); let key = TableScopedPath { table: table_ref.clone(), - path, + store_path: StoreScopedPath::new(ObjectStoreUrl::local_filesystem(), path), }; // Initially cache is empty @@ -1895,7 +2002,10 @@ mod tests { let table_ref = Some(TableReference::from("table")); let key = TableScopedPath { table: table_ref, - path: table_base, + store_path: StoreScopedPath::new( + ObjectStoreUrl::local_filesystem(), + table_base, + ), }; cache.put(&key, CachedFileList::new(files)); @@ -1939,7 +2049,10 @@ mod tests { let table_ref = Some(TableReference::from("table")); let key = TableScopedPath { table: table_ref, - path: table_base, + store_path: StoreScopedPath::new( + ObjectStoreUrl::local_filesystem(), + table_base, + ), }; cache.put(&key, CachedFileList::new(files)); let result = cache.get(&key).unwrap(); @@ -1973,7 +2086,10 @@ mod tests { let table_ref = Some(TableReference::from("table")); let key = TableScopedPath { table: table_ref, - path: table_base, + store_path: StoreScopedPath::new( + ObjectStoreUrl::local_filesystem(), + table_base, + ), }; cache.put(&key, CachedFileList::new(files)); let result = cache.get(&key).unwrap(); diff --git a/datafusion/execution/src/cache/mod.rs b/datafusion/execution/src/cache/mod.rs index 2eba533e04b67..b620892bb1dc6 100644 --- a/datafusion/execution/src/cache/mod.rs +++ b/datafusion/execution/src/cache/mod.rs @@ -28,8 +28,11 @@ use object_store::path::Path; use std::collections::hash_map::DefaultHasher; use std::fmt::{Debug, Display, Formatter}; use std::hash::{Hash, Hasher}; +use std::sync::Arc; use std::time::Duration; +use crate::object_store::ObjectStoreUrl; + /// Base trait for cache implementations with common operations. /// /// This trait provides the fundamental cache operations (`get`, `put`, `remove`, etc.) @@ -143,27 +146,87 @@ impl CacheKey for TableScopedPath { } } -/// Each entry is scoped to its use within a specific table so that the cache -/// can differentiate between identical paths in different tables, and +/// Identifies an object within its registered object store. +#[derive(PartialEq, Eq, Hash, Clone, Debug)] +pub struct StoreScopedPath { + /// URL identifying the registered object store. + pub object_store_url: Arc, + /// Location relative to that store. + pub path: Path, +} + +impl StoreScopedPath { + /// Create a cache key for a path in the given object store. + pub fn new(object_store_url: impl Into>, path: Path) -> Self { + Self { + object_store_url: object_store_url.into(), + path, + } + } +} + +impl CacheKey for StoreScopedPath { + fn size(&self) -> usize { + self.heap_size(&mut DFHeapSizeCtx::default()) + } + + fn table_ref(&self) -> Option<&TableReference> { + None + } +} + +impl DFHeapSize for StoreScopedPath { + fn heap_size(&self, ctx: &mut DFHeapSizeCtx) -> usize { + self.object_store_url.as_str().heap_size(ctx) + self.path.as_ref().heap_size(ctx) + } +} + +impl Display for StoreScopedPath { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "{}{}", self.object_store_url, self.path) + } +} + +/// Scopes a path to its object store and optional table, supporting /// table-level cache invalidation. #[derive(PartialEq, Eq, Hash, Clone, Debug)] pub struct TableScopedPath { pub table: Option, - pub path: Path, + /// Object location, including its registered store. + pub store_path: StoreScopedPath, +} + +impl TableScopedPath { + /// Create a cache key for a path in a store and optional table. + pub fn new( + table: Option, + object_store_url: impl Into>, + path: Path, + ) -> Self { + Self { + table, + store_path: StoreScopedPath::new(object_store_url, path), + } + } + + /// Return the path relative to the object store. + pub fn path(&self) -> &Path { + &self.store_path.path + } } impl DFHeapSize for TableScopedPath { fn heap_size(&self, ctx: &mut DFHeapSizeCtx) -> usize { - self.path.as_ref().heap_size(ctx) + self.table.heap_size(ctx) + self.store_path.heap_size(ctx) + self.table.heap_size(ctx) } } impl Display for TableScopedPath { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { if let Some(table) = &self.table { - write!(f, "{}, {}", self.path, table) + write!(f, "{}, {}", self.store_path, table) } else { - write!(f, "{}", self.path) + Display::fmt(&self.store_path, f) } } } @@ -274,3 +337,41 @@ mod schema_fingerprint_tests { assert_eq!(plain, schema_md, "schema metadata must be ignored"); } } + +#[cfg(test)] +mod path_tests { + use super::*; + + #[test] + fn scoped_paths_preserve_store_identity_and_table_scope() { + let store_path = StoreScopedPath::new( + ObjectStoreUrl::parse("s3://bucket").unwrap(), + Path::from("data.parquet"), + ); + assert_eq!(store_path.to_string(), "s3://bucket/data.parquet"); + assert_eq!(store_path.size(), "s3://bucket/data.parquet".len()); + assert_eq!(store_path.table_ref(), None); + + let unscoped = TableScopedPath { + table: None, + store_path: store_path.clone(), + }; + let table = TableReference::bare("example"); + let scoped = TableScopedPath { + table: Some(table.clone()), + store_path, + }; + assert_eq!(unscoped.to_string(), "s3://bucket/data.parquet"); + assert_eq!(scoped.to_string(), "s3://bucket/data.parquet, example"); + assert_eq!(scoped.table_ref(), Some(&table)); + assert_eq!( + scoped.size(), + unscoped.size() + 2 * size_of::() + "example".len() + ); + + // Entries for the same object can coexist inside and outside a table. + let entries = HashMap::from([(unscoped.clone(), 1), (scoped.clone(), 2)]); + assert_eq!(entries.get(&unscoped), Some(&1)); + assert_eq!(entries.get(&scoped), Some(&2)); + } +} diff --git a/datafusion/execution/src/runtime_env.rs b/datafusion/execution/src/runtime_env.rs index fcfe51267e65f..5dee2299caf8f 100644 --- a/datafusion/execution/src/runtime_env.rs +++ b/datafusion/execution/src/runtime_env.rs @@ -208,13 +208,16 @@ impl RuntimeEnv { url: &Url, object_store: Arc, ) -> Option> { + self.cache_manager.invalidate_object_store(url); self.object_store_registry.register_store(url, object_store) } /// Deregisters a custom `ObjectStore` previously registered for a specific url. /// See [`ObjectStoreRegistry::deregister_store`] for more details. pub fn deregister_object_store(&self, url: &Url) -> Result> { - self.object_store_registry.deregister_store(url) + let store = self.object_store_registry.deregister_store(url)?; + self.cache_manager.invalidate_object_store(url); + Ok(store) } /// Retrieves a `ObjectStore` instance for a url by consulting the diff --git a/docs/source/library-user-guide/upgrading/56.0.0.md b/docs/source/library-user-guide/upgrading/56.0.0.md index e927d3ec4ae0e..e2e8c83fabbcd 100644 --- a/docs/source/library-user-guide/upgrading/56.0.0.md +++ b/docs/source/library-user-guide/upgrading/56.0.0.md @@ -1342,3 +1342,43 @@ without declaring one, use `FileScanConfigBuilder::with_preserve_order(true)`. Code that reads `FileScanConfig::preserve_order` to decide whether files may be reordered should call `FileScanConfig::preserves_file_order()` instead, which also accounts for the output ordering. + +### File caches include the object store URL + +File listing, statistics, and Parquet metadata caches now include the registered +object store URL in their keys. This prevents identical paths in different buckets +from sharing cached results. Statistics and Parquet metadata validation requires matching size and +modification time, plus strong ETags when both metadata values provide them, +otherwise versions when both provide them. Weak HTTP ETags are treated +as missing. ETags remain opaque: differences such as quoted versus unquoted values +cause a cache miss. + +`RuntimeEnv::register_object_store` and `deregister_object_store` invalidate +cached entries for that registry URL, including entries for named and anonymous +tables. Other stores retain their entries. + +`FileFormat` inference signatures are unchanged. Formats can opt into a scoped +metadata cache through the default `with_file_metadata_cache` method. Parquet +inference without a configured scope reads fresh metadata. + +**Migration guide:** + +- `TableScopedPath` contains a `store_path: StoreScopedPath` field alongside its + optional `table`, replacing the previous `path` field. Use + `TableScopedPath::new(table, object_store_url, path)` to construct a key and + `path()` to access its relative path. Store URLs are shared through `Arc`. +- `FileMetadataCache` uses `StoreScopedPath` instead of `object_store::path::Path` + as its key. Construct keys with + `datafusion_execution::cache::StoreScopedPath::new(object_store_url, path)`. +- `CachedParquetFileReaderFactory::new` now takes + `(store, object_store_url, metadata_cache)`. +- `DFParquetMetadata::with_file_metadata_cache` and + `ParquetFileReader::with_metadata_cache` take + `Option<(Arc, Arc)>`. Use `Some((cache, url))` + to enable scoped caching and `None` to disable it. + +The CLI's `metadata_cache()`, `statistics_cache()`, and `list_files_cache()` +functions append an `object_store_url` column, preserving the existing column +positions. Their `path` columns remain relative to that store. + +See [issue #25841](https://github.com/apache/datafusion/issues/25841) for details. diff --git a/docs/source/user-guide/cli/functions.md b/docs/source/user-guide/cli/functions.md index eb7f70e66a1a5..5b1f9e27b3ab7 100644 --- a/docs/source/user-guide/cli/functions.md +++ b/docs/source/user-guide/cli/functions.md @@ -150,6 +150,8 @@ The columns of the returned table are: The `metadata_cache` function shows information about the default File Metadata Cache that is used by the [`ListingTable`] implementation in DataFusion. This cache is used to speed up reading metadata from files when scanning directories with many files. +The `path` column is relative to the object store; `object_store_url` identifies +the store. Together they distinguish files with the same path in different stores. For example, after creating a table with the [CREATE EXTERNAL TABLE](../sql/ddl.md#create-external-table) command: @@ -164,15 +166,14 @@ You can inspect the metadata cache by querying the `metadata_cache` function: ```sql > select * from metadata_cache(); -+----------------------------------------------------+---------------------+-----------------+---------------------------------------+---------+---------------------+------+------------------+ -| path | file_modified | file_size_bytes | e_tag | version | metadata_size_bytes | hits | extra | -+----------------------------------------------------+---------------------+-----------------+---------------------------------------+---------+---------------------+------+------------------+ -| hits_compatible/athena_partitioned/hits_61.parquet | 2022-07-03T15:40:34 | 117270944 | "5db11cad1ca0d80d748fc92c914b010a-6" | NULL | 212949 | 0 | page_index=false | -| hits_compatible/athena_partitioned/hits_32.parquet | 2022-07-03T15:37:17 | 94506004 | "2f7db49a9fe242179590b615b94a39d2-5" | NULL | 278157 | 0 | page_index=false | -| hits_compatible/athena_partitioned/hits_40.parquet | 2022-07-03T15:38:07 | 142508647 | "9e5852b45a469d5a05bf270a286eab8a-8" | NULL | 212917 | 0 | page_index=false | -| hits_compatible/athena_partitioned/hits_93.parquet | 2022-07-03T15:44:07 | 127987774 | "751100bf0dac7d489b9836abf3108b99-7" | NULL | 278318 | 0 | page_index=false | -| . | -+----------------------------------------------------+---------------------+-----------------+---------------------------------------+---------+---------------------+------+------------------+ ++----------------------------------------------------+---------------------+-----------------+--------------------------------------+---------+---------------------+------+------------------+----------------------------------+ +| path | file_modified | file_size_bytes | e_tag | version | metadata_size_bytes | hits | extra | object_store_url | ++----------------------------------------------------+---------------------+-----------------+--------------------------------------+---------+---------------------+------+------------------+----------------------------------+ +| hits_compatible/athena_partitioned/hits_61.parquet | 2022-07-03T15:40:34 | 117270944 | "5db11cad1ca0d80d748fc92c914b010a-6" | NULL | 212949 | 0 | page_index=false | s3://clickhouse-public-datasets/ | +| hits_compatible/athena_partitioned/hits_32.parquet | 2022-07-03T15:37:17 | 94506004 | "2f7db49a9fe242179590b615b94a39d2-5" | NULL | 278157 | 0 | page_index=false | s3://clickhouse-public-datasets/ | +| hits_compatible/athena_partitioned/hits_40.parquet | 2022-07-03T15:38:07 | 142508647 | "9e5852b45a469d5a05bf270a286eab8a-8" | NULL | 212917 | 0 | page_index=false | s3://clickhouse-public-datasets/ | +| hits_compatible/athena_partitioned/hits_93.parquet | 2022-07-03T15:44:07 | 127987774 | "751100bf0dac7d489b9836abf3108b99-7" | NULL | 278318 | 0 | page_index=false | s3://clickhouse-public-datasets/ | ++----------------------------------------------------+---------------------+-----------------+--------------------------------------+---------+---------------------+------+------------------+----------------------------------+ ``` Since `metadata_cache` is a normal table function, you can use it in most places you can use @@ -201,6 +202,7 @@ The columns of the returned table are: | metadata_size_bytes | UInt64 | Size of the cached metadata in memory (not its thrift encoded form) | | hits | UInt64 | Number of times the cached metadata has been accessed | | extra | Utf8 | Extra information about the cached metadata (e.g., if page index information is included) | +| object_store_url | Utf8 | URL identifying the object store, such as `s3://bucket/` or `file:///` | ## `statistics_cache` @@ -213,11 +215,11 @@ You can inspect the statistics cache by querying the `statistics_cache` function ```sql > select * from statistics_cache(); -+------------------+---------------------+-----------------+------------------------+---------+-----------------+-------------+--------------------+-----------------------+ -| path | file_modified | file_size_bytes | e_tag | version | num_rows | num_columns | table_size_bytes | statistics_size_bytes | -+------------------+---------------------+-----------------+------------------------+---------+-----------------+-------------+--------------------+-----------------------+ -| .../hits.parquet | 2022-06-25T22:22:22 | 14779976446 | 0-5e24d1ee16380-370f48 | NULL | Exact(99997497) | 105 | Exact(36445943240) | 0 | -+------------------+---------------------+-----------------+------------------------+---------+-----------------+-------------+--------------------+-----------------------+ ++--------------+-------+---------------------+-----------------+------------------------+---------+-----------------+-------------+--------------------+-----------------------+------+----------------------------------+ +| path | table | file_modified | file_size_bytes | e_tag | version | num_rows | num_columns | table_size_bytes | statistics_size_bytes | hits | object_store_url | ++--------------+-------+---------------------+-----------------+------------------------+---------+-----------------+-------------+--------------------+-----------------------+------+----------------------------------+ +| hits.parquet | hits | 2022-06-25T22:22:22 | 14779976446 | 0-5e24d1ee16380-370f48 | NULL | Exact(99997497) | 105 | Exact(36445943240) | 0 | 0 | s3://clickhouse-public-datasets/ | ++--------------+-------+---------------------+-----------------+------------------------+---------+-----------------+-------------+--------------------+-----------------------+------+----------------------------------+ ``` The columns of the returned table are: @@ -225,6 +227,7 @@ The columns of the returned table are: | column_name | data_type | Description | | --------------------- | --------- | ---------------------------------------------------------------------------- | | path | Utf8 | File path relative to the object store / filesystem root | +| table | Utf8 | Name of the table, or an empty string for an anonymous read | | file_modified | Timestamp | Last modified time of the file | | file_size_bytes | UInt64 | Size of the file in bytes | | e_tag | Utf8 | [Entity Tag] (ETag) of the file if available | @@ -234,10 +237,11 @@ The columns of the returned table are: | table_size_bytes | Utf8 | Size of the table, in bytes | | hits | UInt64 | Number of times the cached file statistics has been accessed | | statistics_size_bytes | UInt64 | Size of the cached statistics in memory | +| object_store_url | Utf8 | URL identifying the object store, such as `s3://bucket/` or `file:///` | ## `list_files_cache` -The `list_files_cache` function shows information about the `ListFilesCache` that is used by the [`ListingTable`] implementation in DataFusion. When creating a [`ListingTable`], DataFusion lists the files in the table's location and caches results in the `ListFilesCache`. Subsequent queries against the same table can reuse this cached information instead of re-listing the files. Cache entries are scoped to tables. +The `list_files_cache` function shows information about the `ListFilesCache` that is used by the [`ListingTable`] implementation in DataFusion. When creating a [`ListingTable`], DataFusion lists the files in the table's location and caches results in the `ListFilesCache`. Subsequent queries against the same table can reuse this cached information instead of re-listing the files. Cache entries are scoped to the object store and, when available, the table. You can inspect the cache by querying the `list_files_cache` function. For example, @@ -247,33 +251,34 @@ You can inspect the cache by querying the `list_files_cache` function. For examp stored as parquet location 's3://overturemaps-us-west-2/release/2025-12-17.0/theme=base/type=infrastructure'; 0 row(s) fetched. -> select table, path, metadata_size_bytes, expires_in, unnest(metadata_list)['file_size_bytes'] as file_size_bytes, unnest(metadata_list)['e_tag'] as e_tag from list_files_cache() limit 10; -+--------------+-----------------------------------------------------+---------------------+-----------------------------------+-----------------+---------------------------------------+ -| table | path | metadata_size_bytes | expires_in | file_size_bytes | e_tag | -+--------------+-----------------------------------------------------+---------------------+-----------------------------------+-----------------+---------------------------------------+ -| overturemaps | release/2025-12-17.0/theme=base/type=infrastructure | 2750 | 0 days 0 hours 0 mins 25.264 secs | 999055952 | "35fc8fbe8400960b54c66fbb408c48e8-60" | -| overturemaps | release/2025-12-17.0/theme=base/type=infrastructure | 2750 | 0 days 0 hours 0 mins 25.264 secs | 975592768 | "8a16e10b722681cdc00242564b502965-59" | -| overturemaps | release/2025-12-17.0/theme=base/type=infrastructure | 2750 | 0 days 0 hours 0 mins 25.264 secs | 1082925747 | "24cd13ddb5e0e438952d2499f5dabe06-65" | -| overturemaps | release/2025-12-17.0/theme=base/type=infrastructure | 2750 | 0 days 0 hours 0 mins 25.264 secs | 1008425557 | "37663e31c7c64d4ef355882bcd47e361-61" | -| overturemaps | release/2025-12-17.0/theme=base/type=infrastructure | 2750 | 0 days 0 hours 0 mins 25.264 secs | 1065561905 | "4e7c50d2d1b3c5ed7b82b4898f5ac332-64" | -| overturemaps | release/2025-12-17.0/theme=base/type=infrastructure | 2750 | 0 days 0 hours 0 mins 25.264 secs | 1045655427 | "8fff7e6a72d375eba668727c55d4f103-63" | -| overturemaps | release/2025-12-17.0/theme=base/type=infrastructure | 2750 | 0 days 0 hours 0 mins 25.264 secs | 1086822683 | "b67167d8022d778936c330a52a5f1922-65" | -| overturemaps | release/2025-12-17.0/theme=base/type=infrastructure | 2750 | 0 days 0 hours 0 mins 25.264 secs | 1016732378 | "6d70857a0473ed9ed3fc6e149814168b-61" | -| overturemaps | release/2025-12-17.0/theme=base/type=infrastructure | 2750 | 0 days 0 hours 0 mins 25.264 secs | 991363784 | "c9cafb42fcbb413f851691c895dd7c2b-60" | -| overturemaps | release/2025-12-17.0/theme=base/type=infrastructure | 2750 | 0 days 0 hours 0 mins 25.264 secs | 1032469715 | "7540252d0d67158297a67038a3365e0f-62" | -+--------------+-----------------------------------------------------+---------------------+-----------------------------------+-----------------+---------------------------------------+ +> select table, path, object_store_url, metadata_size_bytes, expires_in, unnest(metadata_list)['file_size_bytes'] as file_size_bytes, unnest(metadata_list)['e_tag'] as e_tag from list_files_cache() limit 10; ++--------------+-----------------------------------------------------+------------------------------+---------------------+-----------------------------------+-----------------+---------------------------------------+ +| table | path | object_store_url | metadata_size_bytes | expires_in | file_size_bytes | e_tag | ++--------------+-----------------------------------------------------+------------------------------+---------------------+-----------------------------------+-----------------+---------------------------------------+ +| overturemaps | release/2025-12-17.0/theme=base/type=infrastructure | s3://overturemaps-us-west-2/ | 2750 | 0 days 0 hours 0 mins 25.264 secs | 999055952 | "35fc8fbe8400960b54c66fbb408c48e8-60" | +| overturemaps | release/2025-12-17.0/theme=base/type=infrastructure | s3://overturemaps-us-west-2/ | 2750 | 0 days 0 hours 0 mins 25.264 secs | 975592768 | "8a16e10b722681cdc00242564b502965-59" | +| overturemaps | release/2025-12-17.0/theme=base/type=infrastructure | s3://overturemaps-us-west-2/ | 2750 | 0 days 0 hours 0 mins 25.264 secs | 1082925747 | "24cd13ddb5e0e438952d2499f5dabe06-65" | +| overturemaps | release/2025-12-17.0/theme=base/type=infrastructure | s3://overturemaps-us-west-2/ | 2750 | 0 days 0 hours 0 mins 25.264 secs | 1008425557 | "37663e31c7c64d4ef355882bcd47e361-61" | +| overturemaps | release/2025-12-17.0/theme=base/type=infrastructure | s3://overturemaps-us-west-2/ | 2750 | 0 days 0 hours 0 mins 25.264 secs | 1065561905 | "4e7c50d2d1b3c5ed7b82b4898f5ac332-64" | +| overturemaps | release/2025-12-17.0/theme=base/type=infrastructure | s3://overturemaps-us-west-2/ | 2750 | 0 days 0 hours 0 mins 25.264 secs | 1045655427 | "8fff7e6a72d375eba668727c55d4f103-63" | +| overturemaps | release/2025-12-17.0/theme=base/type=infrastructure | s3://overturemaps-us-west-2/ | 2750 | 0 days 0 hours 0 mins 25.264 secs | 1086822683 | "b67167d8022d778936c330a52a5f1922-65" | +| overturemaps | release/2025-12-17.0/theme=base/type=infrastructure | s3://overturemaps-us-west-2/ | 2750 | 0 days 0 hours 0 mins 25.264 secs | 1016732378 | "6d70857a0473ed9ed3fc6e149814168b-61" | +| overturemaps | release/2025-12-17.0/theme=base/type=infrastructure | s3://overturemaps-us-west-2/ | 2750 | 0 days 0 hours 0 mins 25.264 secs | 991363784 | "c9cafb42fcbb413f851691c895dd7c2b-60" | +| overturemaps | release/2025-12-17.0/theme=base/type=infrastructure | s3://overturemaps-us-west-2/ | 2750 | 0 days 0 hours 0 mins 25.264 secs | 1032469715 | "7540252d0d67158297a67038a3365e0f-62" | ++--------------+-----------------------------------------------------+------------------------------+---------------------+-----------------------------------+-----------------+---------------------------------------+ ``` The columns of the returned table are: -| column_name | data_type | Description | -| ------------------- | ------------ | ------------------------------------------------------------------- | -| table | Utf8 | Name of the table | -| path | Utf8 | File path relative to the object store / filesystem root | -| metadata_size_bytes | UInt64 | Size of the cached metadata in memory (not its thrift encoded form) | -| expires_in | Duration(ms) | Last modified time of the file | -| hits | UInt64 | Number of times the cached metadata has been accessed | -| metadata_list | List(Struct) | List of metadatas, one for each file under the path. | +| column_name | data_type | Description | +| ------------------- | ------------ | ---------------------------------------------------------------------- | +| table | Utf8 | Name of the table | +| path | Utf8 | File path relative to the object store / filesystem root | +| metadata_size_bytes | UInt64 | Size of the cached metadata in memory (not its thrift encoded form) | +| expires_in | Duration(ms) | Last modified time of the file | +| hits | UInt64 | Number of times the cached metadata has been accessed | +| metadata_list | List(Struct) | List of metadatas, one for each file under the path. | +| object_store_url | Utf8 | URL identifying the object store, such as `s3://bucket/` or `file:///` | A metadata struct in the metadata_list contains the following fields: