andygrove opened a new pull request, #1995: URL: https://github.com/apache/datafusion-ballista/pull/1995
# Which issue does this PR close? Closes #1994. # Rationale for this change On the executor, every task builds a brand-new `RuntimeEnv` from scratch. The default runtime producer runs once per task, so each `TaskContext` gets a fresh `DiskManager`, `CacheManager`, `ObjectStoreRegistry`, and memory pool. Nothing runtime-scoped is shared between tasks on an executor — including tasks of the same session running the same tables. This rebuilds read-side state per task and per query: - **Object-store clients (connection pools, credentials, TLS) are reconstructed on every task.** This is the same class of problem Comet hit on EKS: creating a fresh object store per file read produced ~5,000 DNS queries/sec/pod (~500× vanilla Spark), exceeding the Route 53 Resolver per-ENI limit of 1,024 qps and causing intermittent connection failures. Comet fixed it with a process-wide object-store cache, which collapsed DNS volume back to Spark levels ([writeup](https://datafusion.apache.org/blog/2026/06/03/comet-eks/#excessive-dns-queries), apache/datafusion-comet#3802). Ballista rebuilds the object store even more aggressively — once per task rather than once per file — so it is exposed to the same failure mode against remote object stores at scale. - **Parquet footers are re-parsed on every scan.** A benchmark session that runs all 22 TPC-H queries re-parses the same `lineitem`/`orders` footers on every query, because each task gets its own empty `CacheManager`. This PR shares read-side runtime state across a session's tasks — and across queries within a session — without changing the per-task memory model. # What changes are included in this PR? A bounded, session-keyed cache of shared *base* `RuntimeEnv`s on the executor: - **`SessionRuntimeCache` (new `ballista/executor/src/runtime_cache.rs`)** — an LRU (default 16 sessions) of base `RuntimeEnv`s keyed by `session_id`. A base env carries the shared read-side state: the `ObjectStoreRegistry`, the `CacheManager` (whose Parquet footer / file-metadata cache is thereby reused across the session), and the `DiskManager`. The base env is built outside the lock on a miss, so a miss never stalls other sessions' lookups. - **Per-task memory pool is unchanged.** Each task's real runtime is derived from the shared base via `RuntimeEnvBuilder::from_runtime_env(base).with_memory_pool(fresh_pool)`. `from_runtime_env` reuses the base's inner file-metadata cache and object-store registry while installing a fresh per-task memory pool, so memory isolation is byte-for-byte identical to before. The existing `--memory-pool-size` `FairSpillPool` behavior is preserved (the memory-pool logic is refactored into a `MemoryPoolPolicy` applied on top of the shared base). - **`Executor::produce_runtime_for_session(session_id, config)`** — used by both execution-time call sites (`execution_loop.rs` poll path, `executor_server.rs` push path). The decode-time throwaway env in `from_proto::get_task_definition` (which does no I/O) is left unchanged. - **Escape hatch preserved.** Session caching applies only to the default runtime producer. When `override_runtime_producer` is set, the cache is left unset and tasks are served per-task exactly as before. - **New config knob** `session_runtime_cache_capacity` (`ExecutorProcessConfig` + `--session-runtime-cache-capacity`, default 16). `0` disables caching and restores the previous fresh-per-task behavior. Eviction is bounded and correctness-neutral: sessions are held in an LRU with a hard cap, and a wrongly-evicted-then-reused session simply rebuilds its base env. There is no reliable session-end signal at the executor (`RemoveJobData` is job-scoped, and a session spans many jobs), so the LRU cap — not a lifecycle hook — is what bounds memory. Covered by unit tests in the executor crate: same-session sharing of the footer cache + registry through the per-task pool rebuild, cross-session isolation, LRU eviction, capacity-0 passthrough, the override fallback, and preservation of the `FairSpillPool` size. # Are there any user-facing changes? One new, optional executor config: `--session-runtime-cache-capacity` (`ExecutorProcessConfig::session_runtime_cache_capacity`), default `16`; `0` restores the previous behavior of building a fresh runtime per task. No breaking API changes; no change to the per-task memory model or to query results. -- 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]
