andygrove opened a new pull request, #4995: URL: https://github.com/apache/datafusion-comet/pull/4995
## 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 `FileMetadataCache` from the task's `RuntimeEnv`, and Comet builds a fresh `RuntimeEnv` for 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 builds `PartitionedFile` with `PartitionedFile::new_with_range`, which hardcodes `last_modified` to epoch 0, and `SparkPartitionedFile` carried 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: - `SparkPartitionedFile` gains a `modification_time` field, populated from Spark's `PartitionedFile.modificationTime` in `partition2Proto`. - `PhysicalPlanner::get_partitioned_files` sets `object_meta.last_modified` from it, so DataFusion's `(size, last_modified)` validation is real. Share the cache: - New `native/core/src/parquet/metadata_cache.rs` holds a process-wide registry of `FileMetadataCache` instances, keyed by object store URL. It parallels the existing process-wide `object_store_cache` in `parquet_support.rs`, including its rationale for process lifetime. Keying per object store matters because `DefaultFilesMetadataCache` keys entries on a bare `object_store::path::Path`, and Comet builds that path from `url.path()`, which drops the bucket. - `init_datasource_exec` uses the registry, falling back to the per-task `RuntimeEnv` cache when sharing is disabled. Configuration and docs: - `spark.comet.scan.metadataCache.enabled` (default `true`) and `spark.comet.scan.metadataCache.memoryLimit` (default 50MB, matching DataFusion's own default). Both are injected explicitly in `serializeCometSQLConfs`, since `SQLConf.getAllConfs` carries only explicitly-set values. - A tuning guide section covering the memory scope, eviction, and validation. 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 returns `None`, the configured limit reaches new and live caches, and concurrent callers on one URL all receive the same instance. - `parquet_exec.rs`: two independent `SessionContext`s (standing in for two tasks) over one file, asserting the second sees a cache hit through the entry's hit counter; and a file whose `last_modified` advances, asserting the cached entry is replaced rather than served stale. The existing `caches_full_metadata_with_page_index` regression test now targets the shared cache. - `planner.rs`: the modification time from the protobuf reaches `object_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: `make` build, `cargo clippy --all-targets --workspace -- -D warnings`, `cargo fmt`, `spotless:check`, the native `parquet` (92) and `planner` (16) test modules, and `ParquetReadV1Suite` (51). -- 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]
