[andygrove]

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 (Spark application, stage id, stage attempt). Tasks of the same stage attempt on an executor share one cache. The registry holds a weak reference to every cache, which the running tasks own through their RuntimeEnv, and strong references to the last two it handed out, so an executor that runs one task at a time still gets hits and at most two caches outlive their tasks. A stage's cache holds DataFusion's 50 MiB per-task metadata limit for each of its running tasks, so sharing leaves every running task the headroom it had with a cache of its own. Each task holds a TaskCache handle, which counts it in while its plan runs and out when the plan drops it. A cache with no running task shrinks to 50 MiB. The handle ignores the limit a task's CacheManager would set, so a new task cannot reset the stage's limit.
• Each scan uses a view of that cache that also keys entries by the scan's registered object store URL, which carries the backend and a hash of the store configuration. DataFusion keys entries by path alone, and the path of a scanned file has no bucket, so equal-size files at the same path in two stores would otherwise share a footer. The Delta deletion vector planning reads footers through the same view.
• 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 application id, stage id and attempt through the existing serialized-config channel, so there is no JNI signature change. The application id is part of the key because stage ids restart at zero in every SparkContext and the registry outlives one that stops in the same JVM.
• A new config, spark.comet.parquet.metadataCache.sharedWithinStage.enabled (default true), turns it off, which restores the per-task cache. Its doc and the accounting-gap section of memory_management.md say what the cache holds and that up to two caches stay resident after their tasks finish.
• 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?

• Rust unit tests for the registry: tasks of a stage share a cache, stages, attempts and applications do not, an idle cache is kept for the next task, idle caches are bounded without dropping running ones, retention is bounded once every task has finished, a stage's cache holds one task's limit per running task and shrinks to one when idle, and building a task's RuntimeEnv does not reset that limit.

• A scan-level Rust test (equal_paths_in_two_stores_do_not_share_footers) reads equal-size files at the same path in two stores with a predicate that selects only the second, with the cache shared and not shared. Without sharing, both scans run in one RuntimeEnv. Each half fails without the per-store scoping.

• CometSharedMetadataCacheRestartSuite stops and recreates the local SparkContext, replaces an equal-size file at the same path, and checks the new context reads the new contents. It fails without the application id in the key.

• 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" reads a relation natively, deletes one of its files, and expects the second read to fail on the missing file.

• Ran locally on Spark 4.1 and 3.5: the new CometExecSuite tests, the perf: load the next Parquet file's footer and page index while the current file is read #7094 prefetch test and CometSharedMetadataCacheRestartSuite pass, and the cache Rust tests pass. cargo clippy -D warnings and make format are clean.

• Hit rate at the default 128 MB split size, measured with Spark 4.1 in local-cluster mode on one Linux box with a release build, over TPC-H lineitem, three runs each:

  Data	Executors x cores	Tasks	Footer reads, sharing on	Footer reads, sharing off
  SF100, 16 files of 1.47 GB	2 x 8	176	56 to 67	176
  SF100, 16 files of 1.47 GB	4 x 4	176	79 to 106	176
  SF10, one file of 2.4 GB	2 x 8 and 4 x 4	19	16	19

  Wall time doesn't move on local NVMe (about 0.8 s for SF100), where a footer read is cheap. The saving is the footer and page index requests to an object store, which this setup can't show.

• Not run locally: HiveMetadataCacheSuite in sql_hive-1, the shard that failed feat: share Parquet metadata cache across tasks in an executor #4995. It needs the run-spark-4.1-hive-tests label, and run-spark-3.5-tests for the Spark 3.5 failure.

[rich7420]

The stage-attempt scope makes sense. I found three identity/lifetime issues to address below. Reviewed 8f0a5cf independently with Claude, then checked the findings against the code and two standalone Rust probes using DataFusion 55.2 / Parquet 59.3. These are component reproductions, not full Spark query runs. The current-head CI passed both new tests and Spark 4.1 HiveMetadataCacheSuite.

[rich7420]

Could we scope this cache by the resolved object-store identity too? DataFusion keys it by ObjectMeta.location, but get_partitioned_files strips the bucket/authority and sets the modification time to the epoch. Tasks reading equal-size files at the same path in different buckets can therefore reuse each other's footer.

I reproduced this with two stores containing table/part-00000.parquet, both 284 bytes: A contains [1,2,3], B [101,102,103]. After warming A, B receives A's exact footer Arc and min/max 1..3; without sharing B reports 101..103. A filter can then prune valid rows.

Please resolve the cache per scan using the existing store identity, including backend/config distinctions, at both parquet_exec.rs:230 and the footer lookup in delta_dv.rs:683. Passing modification time alone does not establish identity. Could we add a native UNION regression over same-path, same-size files in two stores, comparing sharing on/off with a predicate that selects only B?

[andygrove]

Good catch. DataFusion keys the cache by ObjectMeta.location, and the location of a scanned file has no bucket, so equal-size files at the same path in two stores passed the size check and shared a footer.

Each scan now looks up footers through a view of the cache that also keys by the scan's registered object-store URL (scope_to_store). That URL already carries the backend and a hash of the store config, so I didn't add a separate identity. It's applied at parquet_exec.rs:230 and in the Delta deletion-vector footer read, which gets the scan's URL through attach_access_plans so both find the same entries. The view is used whether sharing is on or off.

The regression is equal_paths_in_two_stores_do_not_share_footers. It puts equal-size files holding [1,2,3] and [101,102,103] at the same path in two stores and scans each with a > 100, with the cache shared and not shared. It fails without the scoping ([[], []]). It runs two sequential scans rather than a UNION. Is that enough for what you had in mind?

[rich7420]

Could we make the idle bound hold when tasks release their caches, without waiting for another cache_for call? Holding 16 stage handles and then dropping all of them leaves 16 idle entries pinned indefinitely. Each can retain 50 MiB outside Spark's memory accounting.

A smaller ownership model would be a weak-reference registry plus a bounded queue of strong references, refreshed on acquisition, with expired weak entries pruned. That keeps sequential-task reuse and removes the strong_count arithmetic.

This regression can go in the existing test module; it fails with the current registry:

#[test]
fn idle_retention_is_bounded_after_all_tasks_finish() {
    let caches = SharedMetadataCaches::default();
    let handles: Vec<_> = (0..16)
        .map(|stage| caches.cache_for(key(stage, 0)))
        .collect();
    let weak: Vec<_> = handles.iter().map(Arc::downgrade).collect();
    drop(handles);
    assert!(weak.iter().filter(|w| w.upgrade().is_some()).count() <= MAX_IDLE_CACHES);
}

[andygrove]

You're right. The bound was only enforced on the next cache_for call, so dropping 16 handles left 16 caches pinned.

I took the weak-reference model you suggested. The registry now keeps a Weak for every cache, which the running tasks own through their RuntimeEnv, plus a queue of strong references to the last two it handed out, refreshed on acquisition. Expired weak entries are pruned on each request, and the strong_count arithmetic is gone. At most MAX_IDLE_CACHES caches outlive their tasks, and sequential tasks of one stage still get hits.

I added your test as idle_retention_is_bounded_after_all_tasks_finish, plus one that checks dropped caches are forgotten on the next request.

[rich7420]

Could the namespace also include an application/context identity? Spark's DAGScheduler.nextStageId starts at zero for each SparkContext, while this native static registry survives a stop/start in the same local JVM. A new context can match an old retained stage/attempt and reuse stale metadata for the same path and size.

Please send an application/context identifier alongside the stage fields in serializeCometSQLConfs and include it in this key. A regression should stop/recreate the local SparkContext, reuse the stage ID and an equal-size replaced file, and verify the new contents. This finding is source-traced; I have not run that full restart scenario.

[andygrove]

Agreed, stage ids restart at zero in every SparkContext and the registry outlives a context stopped in the same JVM.

serializeCometSQLConfs now sends spark.app.id with the stage fields, and it's part of the key. If there's no app id, sharing is off for that task rather than guessing.

I ran the restart scenario. CometSharedMetadataCacheRestartSuite stops and recreates the local context, replaces an equal-size file at the same path, and counts rows with id >= 1000. With the app id taken out of the key it fails with 0 did not equal 1000, because the stale footer prunes every row. With it, the suite passes.

[sunchao]

Reviewed the full 11-file diff from d0f367dc02520fe68f524db22083cc0af6e58bca to c904265490344512b954d2d007b5f6da37419c5e. The PR is not a draft. Applied review-comet-pr, review-comet-ffi-pr, and review-comet-memory-pr.

Summary

• Prior state and problem: Each task created its own RuntimeEnv metadata cache, repeatedly fetching and decoding footers and page indexes when tasks read splits of the same file.
• Design approach: Share DataFusion’s metadata cache within an application/stage/attempt, with weak ownership tracking and two retained caches for sequential tasks. Namespace entries by registered object-store identity.
• Correctness: The application, attempt, and store boundaries address the three existing review concerns. Normal scans and Delta footer reads use the same store scope. Registry, cross-store scan, and Delta tests passed. No additional runtime P1/P2 issue was identified. The new CI registration failure is reported below.
• Compatibility analysis: Compared DAGScheduler, TaskContext, ParquetFileFormat, and HiveMetadataCacheSuite sources for Spark 3.4.3, 3.5.9, 4.0.4, 4.1.3, and 4.2.0. The stage APIs are available throughout, and fresh stage attempts preserve the intended boundary for detecting deleted files on later reads. JNI signatures and protobuf schemas remain unchanged.
• Key design decisions: Stage scoping avoids cross-query stale metadata. Application identity handles context restarts. Registered store URLs preserve backend/configuration distinctions. The supported, default-enabled configuration provides an opt-out.
• Implementation sketch: serializeCometSQLConfs supplies task identity, prepare_datafusion_session_context installs the shared cache, and scope_to_store wraps lookups in both Parquet execution and Delta access-plan construction.
• Performance: Sharing avoids repeated metadata reads after cache warm-up. Concurrent cold misses can still fetch independently. Each cache retains DataFusion’s 50 MiB budget, with at most two caches retained after their tasks finish. These allocations remain outside Spark’s memory accounting. Throughput and contention were not benchmarked.
• Design: Reusing DataFusion’s cache and eviction policy keeps the change focused. Weak references let active tasks retain their caches without an explicit stage-completion callback.
• Abstraction & complexity: StoreScopedCache is a narrow key adapter. Its forwarded operations preserve the underlying capacity and eviction behavior. No additional abstraction-related P1/P2 concern was identified.
• Behavioral changes worth calling out: Compared with branch-1.1 at 89c729a9697467e8f8d89e3eadf1797a75e2328c, the touched cache paths intentionally gain stage-level reuse and store namespacing. Disabling sharing restores per-task cache ownership. Encrypted reads continue bypassing metadata caching.
• Suggested improvements: Register CometSharedMetadataCacheRestartSuite in both PR build matrices so preflight succeeds and the restart regression actually runs in CI.

The existing reviews, conversation, inline comments, and threads were inspected. All three previously raised identity/lifetime concerns are addressed in the current code, despite their threads remaining open.

Exact-head CI: run 39167082870 failed Preflight because the restart suite is unregistered. Required Checks consequently failed, and build/Spark/Delta jobs were skipped. CodeQL and Analyze Actions passed.

Validation: 14 focused native tests passed using the exact-head build with --no-default-features --features delta: seven registry tests, one cross-store scan regression, and six Delta access-plan tests. python dev/ci/check-suites.py independently reproduced exit 255. JVM restart/query tests, Spark SQL/Hive suites, HDFS, and real object-store integration were not run locally. Earlier-head CI passes do not validate this head.

[sunchao]

[P2] [P2] Register the restart suite in both CI matrices. This new suite is absent from .github/workflows/pr_build_linux.yml and .github/workflows/pr_build_macos.yml. Adding it causes the mandatory suite-discovery check to exit 255 instead of allowing CI to proceed. At this exact head, that fails Preflight and Required Checks, skips the build and regression jobs, and leaves the restart test without CI coverage. Please add its fully qualified name to an appropriate suite group in both workflows.

Evidence: Locally, python dev/ci/check-suites.py exits 255 with Suite not found in workflow .github/workflows/pr_build_linux.yml: org.apache.comet.exec.CometSharedMetadataCacheRestartSuite. Both workflow files lack the suite name. Exact-head CI reports the same failure: https://github.com/apache/datafusion-comet/actions/runs/39167082870/job/117792161514.

[andygrove]

Thanks, added it to the exec group in both workflows, next to CometExecSuite, in e480116. dev/ci/check-suites.py exits 0 now.

[comphead]

Reusing DataFusion's DefaultCache through the cache-manager hook keeps this small, and the store scope, weak registry and app id from the earlier threads are in. The prefetch from #7094, now on main, loads through the same factory cache, so I do not see the two fetching a footer twice. Four comments inline. Separately, this PR changes delta_dv.rs and delta_spark_scan.rs, and sharing is on by default for Delta scans too, but the Delta contrib jobs were skipped at e480116496. Could run-delta-tests give a verdict here? The merge queue does not run them.

[comphead]

Each task used to get its own 50 MiB DefaultCache. Now every concurrent task of the stage on this executor shares one 50 MiB LRU. The Delta deletion vector path depends on the entry surviving: attach_access_plans reads the footer of every DV file in the partition at planning time, and the comment in attach_access_plan says that fetch "also serves the subsequent data-file open". With one LRU per stage, other tasks' footers and page indexes land in between, and the entries this task has not opened yet are the least recently used. So with enough concurrent tasks and wide files with page indexes, the scan may fetch those footers again. I have not measured how often this happens. Was the 50 MiB meant to cover all concurrent tasks of a stage? Scaling the limit with the executor's task slots would keep the old per-task headroom.

[andygrove]

No, it wasn't meant to, and you're right that it shrinks every task's headroom when the tasks read different files. A stage's cache now holds DEFAULT_METADATA_CACHE_LIMIT for each of its running tasks. Each task takes a TaskCache handle when it builds its RuntimeEnv, which counts it in, and drops it with its plan, which counts it out. So each running task keeps the 50 MiB it had with a cache of its own, and the footers the deletion vector planning loads are no more likely to be evicted before the scan opens them than before. A file the tasks share is stored once, so the tasks of a stage hold no more than they did.

When the last task finishes, the cache shrinks back to 50 MiB, evicting the least recently used entries. So each of the two caches the registry keeps holds no more than a task's own cache would, at most 100 MiB with no task running. (40beedd)

An earlier version scaled the limit by spark.executor.cores / spark.task.cpus, but sunchao pointed out that this is only an estimate on a standalone executor, so the limit now follows the tasks that actually run.

[comphead]

With shared == false each scan builds its own RuntimeEnv inside the inner loop, so the two scans never see the same cache and this half passes without scope_to_store. With sharing off, the collision the view guards against needs two scans in one plan, so one RuntimeEnv. Could the unshared iteration build one RuntimeEnv for both scans? Then the "or private to one task" part of the doc comment is tested too. The registry and stage_cache built for that iteration are also unused.

[andygrove]

Done. Without sharing, both scans now run in one RuntimeEnv with its own cache, as two scans in one plan would. The registry and the stage cache are only built for the shared iteration.

I checked that each half needs the scoping. With scope_to_store returning the cache unscoped, the shared iteration fails, and so does the unshared one when it runs first.

[comphead]

"once per executor" is the best case, since tasks that start together each miss, as the new CometExecSuite test notes. This text becomes the configs.md entry of a default-on setting, so could it also say what the cache holds? Each stage keeps up to 50 MiB of decoded footers and page indexes outside Spark's memory accounting, and up to two such caches outlive their tasks.

[viirya]

Agreed. Could memory_management.md get the same note? Its "Who allocates what" table and the accounting-gap section don't cover native allocations that outlive the task. As far as I know, the two retained caches are the first ones that do.

[andygrove]

Updated. The doc now says that tasks which start together each fetch the footer, so the saving grows with the number of splits per file. It also says what the cache holds: up to 50 MiB of decoded footers and page indexes for each of the stage's running tasks, outside Spark's memory accounting. Up to two caches are kept after their tasks finish, each holding up to 50 MiB. With the setting off, each task's 50 MiB cache is freed when the task ends.

[andygrove]

Added. The accounting-gap section of memory_management.md now has a bullet for native caches that outlive their task. The metadata cache isn't reserved from the pool. A stage's cache holds up to 50 MiB per running task. The two most recent caches stay resident after their tasks finish, at 50 MiB each, so up to 100 MiB can be held with no task running.

[comphead]

idle_caches_are_bounded_but_running_ones_are_not_dropped below already shows that dropped caches leave the registry, since its len() would be 11 otherwise. Could this test go?

[andygrove]

Dropped, thanks. idle_caches_are_bounded_but_running_ones_are_not_dropped covers it.

[viirya]

Thanks for working through the earlier threads. The store scope, the app id in the key and the weak registry all look right to me. The restart suite also runs in the exec job now. I left three comments inline, and one reply on comphead's CometConf thread about memory_management.md. The biggest one is how often sharing hits with default split sizes, because that decides whether this should be on by default.

The branch also conflicts with main in jni_api.rs and CometExecSuite.scala after #7094. With #7094 the prefetch loads through the factory's cache, so it picks up the store scope after the rebase.

[viirya]

I'd like to understand how often this hits outside the benchmark setup. 470 tasks over a 474 MB file is roughly a 1 MB maxPartitionBytes. At the default 128 MB that file is 4 splits.

Spark sorts splits by length with a stable sort, so the splits of one file land in adjacent partitions. TaskSchedulerImpl.resourceOfferSingleTaskSet hands out one task per shuffled offer per round. So those adjacent tasks start on different executors at about the same time. Tasks that do land on the same executor together each miss, since there is no single-flight. So I'd expect hits mainly when splits per file far exceed the number of executors.

Could you share scan_io_metadata_cache_hits and scan_io_metadata_cache_misses at the default split size with more than one executor? If the win is limited to small splits or very large files, would it make sense to ship this off by default, like prefetchNextFileMetadata in #7094? That would also avoid changing the per-task headroom comphead raised for every scan.

[andygrove]

Fair question. I measured it on one Linux box with Spark 4.1 in local-cluster mode, with a release build, at the default 128 MB maxPartitionBytes. The query was sum(l_quantity), sum(l_extendedprice) over TPC-H lineitem, three runs per setting.

Data	Executors x cores	Tasks	Footer reads, sharing on	Footer reads, sharing off
SF100, 16 files of 1.47 GB	2 x 8	176	56 to 67 (62 to 68% hits)	176
SF100, 16 files of 1.47 GB	4 x 4	176	79 to 106 (40 to 55% hits)	176
SF10, one file of 2.4 GB	2 x 8 and 4 x 4	19	16 (3 hits)	19

So you're right about tasks that start together. A file with few splits gets almost nothing, since its tasks run in one wave and each misses. With files of a size common in data lakes, each split into about 11 tasks, and 176 tasks over 16 slots, the later waves find the footer and footer reads drop by 40 to 68%. More executors means fewer hits, as you predicted.

Wall time on this box doesn't move, about 0.8 s either way, because the data is on local NVMe and a footer read costs very little there. The point of the change is the object store case, where each footer and page index read is one or more GET requests with tens of milliseconds of latency. Removing 40 to 68% of those requests from the start of the tasks is the saving, and local disk can't show it. So I'd like to keep it on by default. With the limit now sized by the stage's running tasks, as in the thread on shared_metadata_cache.rs, turning it on no longer takes headroom away from any scan.

[viirya]

One thing to watch if the limit gets scaled, as comphead suggested on shared_metadata_cache.rs. CacheManager::try_new in DataFusion 55.2 calls file_metadata_cache.update_cache_limit(config.metadata_cache_limit) on the cache it is given. This passes CacheManagerConfig::default(), so every task that builds its RuntimeEnv resets the shared cache to 50 MiB. Today that does nothing because both values are 50 MiB. But a larger limit set only in DefaultCache::new would be undone by the next task. So it would need to go through with_metadata_cache_limit here as well.

[andygrove]

Good catch, thanks. The handle a task's CacheManager gets, TaskCache, now ignores update_cache_limit, so the metadata_cache_limit that CacheManager::try_new applies has no effect on the shared cache. The registry alone sets the limit, from the stage's running tasks. a_task_runtime_env_does_not_reset_the_stage_limit builds a RuntimeEnv with the default config over one task's handle while another task is running, and checks that the limit is still two tasks' worth. It then drops the other task and checks that the limit falls to one task's. (40beedd)

[viirya]

Could this test assert that the plan has a CometNativeScanExec, like the restart suite does? If the read falls back to Spark, the test passes without exercising the cache. intercept[SparkException] also accepts any failure. Checking that the cause is the missing-file error would tie it to the case HiveMetadataCacheSuite cares about.

[andygrove]

Done. The test now asserts that the first read's plan has a CometNativeScanExec. It also checks that the second read fails because of the missing file: a FileNotFoundException in the cause chain. That is the cause of FAILED_READ_FILE.FILE_NOT_EXIST on Spark 4.0+, which HiveMetadataCacheSuite checks for, and of Spark 3.x's file-not-found read error. It passes on 4.1 and 3.5.

[sunchao]

Reviewed commit d14249fd94d0fbf69a1035d9d757e226c516d140. New P1/P2 findings are detailed inline.

[sunchao]

[P2] Use actual executor slots when sizing the shared cache. On a standalone host reporting 12 processors, with spark.cores.max=2, spark.task.cpus=1, and no explicit spark.executor.cores, Spark can allocate a two-core executor while numDriverOrExecutorCores returns 12. The new use of that estimate here gives the stage a 600 MiB cache instead of the intended 100 MiB. A bounded reproduction retained 147,254,940 bytes after task handles were dropped, versus 98,169,960 bytes with the correct two-slot limit. This introduces excess native retention outside Spark's accounting, and two such caches can remain idle. Please pass authoritative executor capacity for cache sizing, or use a conservative bound when it is unknown.

Evidence: Spark 4.1.3 source: StandaloneSchedulerBackend passes allocated cores through --cores, Master.scheduleExecutorsOnWorkers caps allocation by app.coresLeft, and CoarseGrainedExecutorBackend copies cores into spark.executor.cores only for non-default resource profiles. SparkEnv.createExecutorEnv receives the allocated count separately. Comet's existing helper at CometExecIterator.scala:1016 returns availableProcessors for standalone when that config is absent. Disposable test review_executor_core_estimate_cache_retention loaded 12 local Parquet files, each with 128 Int64 columns and 128 two-row row groups, through the current scoped cache. After dropping task handles, results were slots=2 limit=104857600 idle_entries=8 idle_metadata_bytes=98169960 and slots=12 limit=629145600 idle_entries=12 idle_metadata_bytes=147254940. The probe passed with cargo test -p datafusion-comet --lib --no-default-features --features delta review_executor_core_estimate_cache_retention -- --nocapture. Probe source and output are retained at /tmp/comet7082-d142-core-estimate-probe.rs and /tmp/comet7082-d142-core-estimate.log. No full standalone application was run.

[andygrove]

You're right, thanks for the reproduction. That count comes from numDriverOrExecutorCores, which is an estimate on a standalone executor without spark.executor.cores, and it shouldn't size memory. Nothing in the task gets the executor's real core count, so the cache no longer uses one.

Each task now holds a TaskCache handle on its stage's cache while its plan runs. Taking the handle counts the task in, and dropping it counts the task out. The cache holds 50 MiB for each task that is running, and 50 MiB once none is. On your two-core executor that gives at most two tasks' worth, 100 MiB, whatever the host's processor count. A retained cache shrinks to 50 MiB, so the two that outlive their tasks hold at most 100 MiB together. a_stage_cache_holds_one_task_limit_per_running_task covers the counting and the shrink. (40beedd)

This is the same idea as the running-task count in #7096, which the fair pool divides by. This one counts the tasks of one stage rather than of the executor, since only those share the cache.

[andygrove]

Thanks @comphead, @viirya and @sunchao. I merged main, which fixes the conflicts in jni_api.rs and CometExecSuite.scala. As viirya noted, the prefetch from #7094 loads through the reader factory's cache, which is the store-scoped view, so the two share entries and keep the store scope. Each inline thread has a reply.

The main changes:

• A stage's cache now holds the 50 MiB per-task limit for each of its running tasks, counted through a per-task handle, and shrinks to 50 MiB when none is running. The handle ignores the limit a task's CacheManager sets, so a new task can't reset it. sunchao showed that sizing it by executor cores overshoots on standalone executors, so it counts tasks instead.
• Sharing stays on by default. The numbers are in the CometConf thread: at the default split size, footer reads drop by 40 to 68% when files have many splits, and the time saved shows on object stores rather than on local disk.
• The config doc and memory_management.md describe what the cache holds and that the retained caches outlive their tasks.

@comphead, I added run-delta-tests so the Delta contrib jobs give a verdict on the delta_dv.rs and delta_spark_scan.rs changes.

[andygrove]

Benchmarked with TPC-DS @ SF1000. With config on, 492s. With config off 557s. 🚀

[andygrove]

Benchmarked with TPC-DS @ SF1000. With config on, 492s. With config off 557s. 🚀

Actually I am seeing a lot of variance in benchmark results. Disregard these results for now

[mbutrovich]

Thanks @andygrove, first pass.

[mbutrovich]

The contributor guide lists state that "needs cleanup or refresh during process lifetime" as a reason to avoid a singleton. As I read it, the retained caches fit that description. Spark gives every submission of a stage a new attempt number (Stage.makeNewStageAttempt), so once a stage attempt's tasks finish, no task can hit its cache again. The registry frees a retained cache only after two other stages ask for one on this executor, or when the process exits. An executor that goes idle after a query keeps up to 100 MiB that nothing can use.

Should release_runtime clear this registry, as it clears FILE_IO_CACHE through clear_file_io_cache? Then the static gets the same teardown at plugin shutdown as the other process-wide native caches. Could the module doc and the new memory_management.md bullet also say when a retained cache is freed? Today they say how many are kept, but not for how long.