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: