feat: share Parquet metadata cache across tasks in an executor#4995
Draft
andygrove wants to merge 10 commits into
Draft
feat: share Parquet metadata cache across tasks in an executor#4995andygrove wants to merge 10 commits into
andygrove wants to merge 10 commits into
Conversation
Hold the config write lock for the whole configure() operation, including the loop that applies the new limit to live caches, so concurrent configure() calls with different limits can no longer race and leave a cache on a stale limit that the idempotency check would then never revisit. Add a test that exercises cache_for() from multiple threads concurrently and asserts they all observe the same cache instance.
cache_for copied config.memory_limit into a local and released the config read guard before acquiring the caches write lock. A concurrent configure() could run entirely in that window, updating config and resyncing every cache already in the map, none of which yet included the cache cache_for was about to insert. The new cache would then be permanently stuck on the old limit, with configure's idempotency short-circuit never revisiting it. Hold the config read guard for the whole of cache_for, including the caches write lock used to insert a new cache, so a concurrent configure blocks until the insert is complete and visible to its cache scan. Acquisition order stays config-before-caches on both paths, so this cannot deadlock with configure.
Fix code review findings on the shared Parquet metadata cache: correct the false claim that keying by object store URL removes all bucket collisions (ABFS containers on one storage account still share a key, since the container is URL userinfo, not host), correct the memory limit doc to state it bounds each object store's cache rather than one process-wide budget, note that the limit only applies when sharing is enabled, note that the enabled flag is process-global and last-writer-wins, note that staleness validation degrades to size-only without a reported modification time, and document the epoch-0 last_modified risk in the dormant single-file JNI path. Add tests covering a read-after-overwrite in the same session and the per-task fallback scan path when cache sharing is disabled.
The overwrite test previously asserted results only around a
overwrite, but write.mode("overwrite") produces new file names, so
those reads never hit the same cache entry and the test exercised
nothing the shared cache is involved in. Read the same files twice
before the overwrite to cover an actual shared-cache hit, and correct
the comment to describe what the overwrite step does and does not
prove. Use checkSparkAnswer so each read is also checked against
vanilla Spark, consistent with the rest of the suite.
Contributor
|
I remember @comphead tried this and we always ran into a Spark SQL test failure where it creates a file, reads it, deletes it, then reads it again. We'd fail for serving the query from metadata (i.e., cache invalidation wasn't managed). |
Member
Author
Yup, I hit the same issue. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Closes #4994. This is item P5 of #4842.
Rationale for this change
Comet's native Parquet scan takes its
FileMetadataCachefrom the task'sRuntimeEnv, and Comet builds a freshRuntimeEnvfor every Spark task, so the cached footer and page index are discarded when the task ends. Two tasks on one executor re-fetch and re-parse the same footer, which costs a round trip to object storage plus a Thrift parse per task. This happens whenever a large file is split across tasks, and again whenever a later stage or query rescans the same table.Sharing the cache across tasks also lowers memory use: the 50MB budget was previously allocated per task, so an executor with 8 task slots could hold 8 budgets at once.
Making the cache outlive a task requires its validity check to work first. DataFusion validates cached metadata against
(size, last_modified), but Comet buildsPartitionedFilewithPartitionedFile::new_with_range, which hardcodeslast_modifiedto epoch 0, andSparkPartitionedFilecarried no modification time. Validation was therefore size-only. That is harmless for a cache that lives for one task and a wrong-results hazard for one that lives for the executor process, so the modification time is plumbed through as part of this change rather than separately.What changes are included in this PR?
Carry the file modification time to the native side:
SparkPartitionedFilegains amodification_timefield, populated from Spark'sPartitionedFile.modificationTimeinpartition2Proto.PhysicalPlanner::get_partitioned_filessetsobject_meta.last_modifiedfrom it, so DataFusion's(size, last_modified)validation is real.Share the cache:
native/core/src/parquet/metadata_cache.rsholds a process-wide registry ofFileMetadataCacheinstances, keyed by object store URL. It parallels the existing process-wideobject_store_cacheinparquet_support.rs, including its rationale for process lifetime. Keying per object store matters becauseDefaultFilesMetadataCachekeys entries on a bareobject_store::path::Path, and Comet builds that path fromurl.path(), which drops the bucket.init_datasource_execuses the registry, falling back to the per-taskRuntimeEnvcache when sharing is disabled.Configuration and docs:
spark.comet.scan.metadataCache.enabled(defaulttrue) andspark.comet.scan.metadataCache.memoryLimit(default 50MB, matching DataFusion's own default). Both are injected explicitly inserializeCometSQLConfs, sinceSQLConf.getAllConfscarries only explicitly-set values.Two known limitations are documented rather than fixed here. The memory limit bounds each object store's cache within an executor, so an executor reading several buckets can hold a multiple of it. And for
abfs://URLs the container is URL userinfo rather than host, so Comet's object store URL key does not separate two containers in one storage account; those entries rely on(size, last_modified)validation. The second limitation stems from a pre-existing issue in the object store cache itself, filed as #4993.How are these changes tested?
New Rust tests:
metadata_cache.rs: the same URL returns one instance, different buckets get separate instances, a disabled registry returnsNone, the configured limit reaches new and live caches, and concurrent callers on one URL all receive the same instance.parquet_exec.rs: two independentSessionContexts (standing in for two tasks) over one file, asserting the second sees a cache hit through the entry's hit counter; and a file whoselast_modifiedadvances, asserting the cached entry is replaced rather than served stale. The existingcaches_full_metadata_with_page_indexregression test now targets the shared cache.planner.rs: the modification time from the protobuf reachesobject_meta.last_modified.New Scala tests:
CometFilePartitionSerdeSuite: the modification time is serialized into the protobuf.CometMetadataCacheSuite: both configs cross JNI with their defaults; a repeated read of the same files exercises a shared-cache hit across queries in one process; and a scan with sharing disabled exercises the per-task fallback path.Also verified locally:
makebuild,cargo clippy --all-targets --workspace -- -D warnings,cargo fmt,spotless:check, the nativeparquet(92) andplanner(16) test modules, andParquetReadV1Suite(51).