Repository navigation
Conversation
github-actions
Bot
added
documentation
|
Thank you for opening this pull request! Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch). Details |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #25858 +/- ##
==========================================
+ Coverage 82.93% 82.95% +0.02%
==========================================
Files 1154 1154
Lines 457933 458638 +705
Branches 457933 458638 +705
==========================================
+ Hits 379786 380463 +677
- Misses 55128 55136 +8
- Partials 23019 23039 +20 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
villebro
left a comment
villebro
left a comment
There was a problem hiding this comment.
If ETag or version are available, I feel we should IMO not need to compare against last_modified or size. WDYT?
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
Thanks @limadog9, this looks like the right approach to me. Putting the object store URL in the cache keys, rather than checking it on the cached value, means files at the same path in different stores get separate entries instead of evicting each other. Sharing one comparison function between both caches, with a table-driven test over every ETag × version combination, is nice too.
The keys are built from the same URL on the planning path (ListingOptions::infer_schema, ListingTable statistics) and the execution path (ParquetFormat::create_physical_plan and the proto ParquetSource path), so metadata cached during planning is still reused at execution.
I left one comment I think should be addressed before merging: datafusion-cli's metadata_cache() output changes, see the comment on Display for ObjectStorePath. The others are nits. I also added some object_store caveats to the existing thread on the ETag rule.
This is an API change. The semver check flags the FileFormat inference methods, CachedParquetFileReaderFactory::new, DFParquetMetadata::with_file_metadata_cache, ParquetFileReader::with_metadata_cache and TableScopedPath, and the upgrade guide covers all of them.
|
Thanks for the reviews, @andygrove and @villebro. I've updated the PR to address your feedback, including the CLI cache output and ETag handling. Thanks for taking another look. |
villebro
left a comment
villebro
left a comment
There was a problem hiding this comment.
Validated cache_manager.rs changes, LGTM. Referring to @andygrove to review other changes.
|
Hi @andygrove, following up on the cache identity fix. The updates for the CLI cache output and ETag handling are in, and Villebro has approved the cache manager changes. Could you take another look at the remaining changes when you have a chance? The latest Actions runs are still |
c6fdf87 to
2ef0653
Compare
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
Thanks for tackling this. Scoping the metadata, statistics and listing caches by object store and validating by ETag is the right direction, and parquet_caches_separate_object_stores and parquet_caches_refresh_changed_etag are good tests. I have one correctness gap and a few smaller points.
1. Re-registering a store under the same URL keeps the old store's entries valid
The caches are keyed on ObjectStoreUrl, not on the store instance. If register_object_store replaces the store for a URL (new credentials, a different endpoint, a repointed mem://), the old entries stay in the cache. If the new store has a file with the same path, size, mtime and ETag, it gets the old store's footer and statistics.
This reproduces on the PR head (2ef06534). The test below fails with the old data (expected min=7 max=9, actual min=1 max=3):
Test (add to cache_object_identity.rs)
#[tokio::test]
async fn parquet_caches_reregistered_object_store() {
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();
// The two stores are indistinguishable by ObjectMeta, ETag included.
assert_eq!(
first.head(&path).await.unwrap(),
second.head(&path).await.unwrap()
);
let ctx = SessionContext::new();
let url = Url::parse("mem://reused").unwrap();
ctx.runtime_env().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
);
// Point the same URL at a different backend holding different bytes.
ctx.runtime_env().register_object_store(&url, second);
let second_result = aggregate_file(&ctx, "mem://reused/data.parquet").await;
assert_batches_eq!(
[
"+-----+-----+-------+",
"| min | max | count |",
"+-----+-----+-------+",
"| 7 | 9 | 3 |",
"+-----+-----+-------+",
],
&second_result
);
}The setup is narrow, since the ETag and size must also match. Still, the PR title says it isolates object stores, and this is the remaining way for one store's cache entries to reach another. I haven't checked the base commit; I'd expect the same stale result there, so this is probably a gap rather than a regression. Possible fixes: evict entries for the URL when register_object_store replaces a store, or include a per-registration identity in the key.
2. A matching ETag skips the size and mtime check (cache_manager.rs, file_metadata_matches)
This is a design question, not a bug I can show. A non-weak ETag match now returns true without looking at size or mtime, and a matching version does the same. That is correct for stores whose ETags are content or version based (S3, GCS, Azure), but it trusts every store to return unique ETags, which the old check did not. Requiring size and mtime to agree as well would cost nothing in correct cases and keep the old safety net. I'm not asking for a change if you'd rather keep this.
3. Public API changes
FileFormat::infer_statsandinfer_orderinggain an&ObjectStoreUrlparameter. Only Parquet uses it (the other formats take_object_store_url), and every downstreamFileFormatimplementation stops compiling. Could the store scope be applied where the cache handle is created instead, so the trait doesn't change?FileMetadataCacheis now keyed onStoreScopedPath, andTableScopedPath.pathbecamestore_path.path. ATableScopedPath::new(...)constructor orpath()accessor would ease migration. The 56.0.0 upgrade guide helps, but source-level compatibility is lost.datafusion-cli:object_store_urlis inserted right afterpathin themetadata_cache()andstatistics_cache()schemas (and the list-files one), which shifts columns for positional consumers. Appending it would keep them working.
4. Minor
- Each cache
getandputbuilds aStoreScopedPath, cloning the URL and path, per file, andlist_with_cachere-parses the URL string twice. Holding anArc<ObjectStoreUrl>computed once would avoid this. I haven't measured it. with_metadata_cacheandwith_file_metadata_cacherequire a URL even when the cache isNone. Passing the pair asOption<(Arc<cache>, ObjectStoreUrl)>would make that invalid state unrepresentable.
Item 1 is the one I'd like resolved or explicitly scoped out before merge. The rest are suggestions.
Scope listing, statistics and Parquet metadata by object store URL and validate size, modification time, ETag/version and schema independently. Invalidate the affected scope when runtime registrations change, while retaining entries for other stores. Preserve FileFormat inference signatures through optional scoped cache configuration. Share store URLs, reuse metadata keys, provide cache-key migration helpers and append the CLI store column to preserve positions. Cover same-URL replacement, deregistration, stale listings, named and anonymous entries, unchanged metadata across stores and CLI column order.
2ef0653 to
177efad
Compare
Which issue does this PR close?
Closes #25841.
Rationale for this change
An overwrite can keep the same file size and modification time while changing its statistics and Parquet footer. Different object stores can also contain the same relative path with identical metadata. Reusing either cache entry can produce incorrect query results.
What changes are included in this PR?
TableScopedPathcomposes the newStoreScopedPathkey.object_store_urlin all three CLI cache functions while keeping theirpathcolumns relative to the store.Validation still depends on current object metadata; cached listings and changes between planning and execution retain the consistency limitations described in the issue.
What is the testing strategy for this PR?
Regression tests cover identifier precedence, weak and unavailable ETags, schema changes, equal-size overwrites at an unchanged timestamp, and separate stores with identical object metadata. DataFrame tests assert correct aggregates and filtered scans. Cache tests verify isolation, reuse, and listing invalidation. A CLI regression test asserts exact relative paths and store URLs in all three cache functions for matching filenames in two stores.
Passed locally:
cargo test --profile ci -p datafusion-execution -p datafusion-datasource -p datafusion-datasource-parquet -p datafusion-cli --lib --bins cache --no-fail-fast(59 tests)cargo test --profile ci -p datafusion -p datafusion-cli --test core_integration datasource::(26 tests)--lib --tests --all-features -- -D warnings) fordatafusion,datafusion-execution,datafusion-catalog-listing,datafusion-datasource,datafusion-datasource-parquet, anddatafusion-cli.cargo fmt --all -- --check./ci/scripts/doc_prettier_check.sh --write --allow-dirtyAre there any user-facing changes?
File caches no longer reuse entries from another object store or from a changed strong ETag/version when fresh metadata is available. CLI cache functions expose the store URL separately from the relative path.
API changes:
FileFormatinference methods now take an object store URL;TableScopedPathcontains aStoreScopedPath;FileMetadataCacheusesStoreScopedPath; and the cached Parquet reader factory/cache builders require the URL. Seedocs/source/library-user-guide/upgrading/56.0.0.mdfor migration details.