andygrove opened a new pull request, #7082: URL: https://github.com/apache/datafusion-comet/pull/7082
## Which issue does this PR close? Part of #4994. It covers a file split across the tasks of one stage. It does not cover reuse across queries, so it does not close the issue. ## Rationale for this change The native Parquet scan caches each file's footer and page index in a `FileMetadataCache` that lives in the task's `RuntimeEnv`. Comet builds a new `RuntimeEnv` for every Spark task, so tasks that read splits of the same file each fetch and parse the same footer, and each loads the page index for all of the file's row groups. With a 474 MB file split into 470 tasks, footer and page-index decoding was about 9% of task samples. The earlier attempt (#4995) used a cache that outlived the query and was closed because `HiveMetadataCacheSuite` failed. That suite reads a table, deletes a data file, reads again, and expects `FILE_NOT_EXIST`. Spark's cached file index still lists the deleted file with its old size and modification time, so validating a cached footer against it passes. A `count(*)` then never opens the file, and the second read returns the old count. See the comment on #4994 for details. This PR scopes the cache to the stage so that cannot happen: every new query is a new stage and opens its files. ## What changes are included in this PR? - `shared_metadata_cache.rs` adds a process-wide registry of metadata caches keyed by (stage id, stage attempt). Tasks of the same stage attempt on an executor share one cache. The registry keeps the caches of stages with a running task, plus the last two idle ones, so an executor that runs one task at a time still gets hits. - `prepare_datafusion_session_context` installs the shared cache in the task's `RuntimeEnv` when the config is on and the stage is known. - `CometExecIterator.serializeCometSQLConfs` sends the stage id and attempt through the existing serialized-config channel, so there is no JNI signature change. - A new config, `spark.comet.parquet.metadataCache.sharedWithinStage.enabled` (default `true`), turns it off, which restores the per-task cache. - `modificationTime` is not plumbed through to `last_modified`. Stage scoping makes it unnecessary, and validation stays size-only as it is today. Not covered: reuse of footers across queries, the second case in #4994. That conflicts with Spark's semantics for the reason above. ## How are these changes tested? - Four Rust unit tests for the registry: tasks of a stage share a cache, stages and attempts do not, an idle cache is kept for the next task, and idle caches are bounded without dropping running ones. - `CometExecSuite`: "Parquet footers are shared by the tasks of a stage" splits one file into many tasks and checks for cache hits with sharing on and none with it off. "Parquet footers are not shared between queries" deletes a file between two reads of a relation and expects a `SparkException` on the second. - Ran locally on the default Spark 4.1 profile: the new `CometExecSuite` tests and "Comet native metrics: scan" pass, and `cargo clippy -D warnings` and spotless are clean. - Not run locally: `HiveMetadataCacheSuite` in `sql_hive-1`, the shard that failed #4995. It needs the `run-spark-4.1-hive-tests` label, and `run-spark-3.5-tests` for the Spark 3.5 failure. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
