Explore Kubernetes in-place pod memory resize (Spark 4.2 ExecutorResizePlugin) for Comet's untracked native memory #6727

What is the problem the feature request solves?

Sizing spark.executor.memoryOverhead for Comet is a guess made before the job runs. The tuning guide says the memory Comet allocates without reserving it (per-batch working memory, Parquet and decompression buffers, allocator fragmentation and retained pages, JVM-side Arrow buffers) has to fit in the overhead, and that executors that outgrow it are killed with exit code 137. Measurements in #4576 found jemalloc's resident memory at 6x to 8x allocated at the median, so the right overhead depends on the workload and is hard to know up front. Operators over-provision every executor for the worst case, or under-provision and lose executors mid-query.

Spark 4.2.0 adds a built-in org.apache.spark.scheduler.cluster.k8s.ExecutorResizePlugin (SPARK-55432) that uses Kubernetes in-place pod resize (beta in 1.33, GA in 1.35) to grow an executor's memory without restarting it. This looks like a good fit for the untracked part of Comet's footprint, and nobody has looked at it from Comet's side yet.

What the Spark plugin does

Read from branch-4.2 of ExecutorResizePlugin.scala:

• It runs in the driver, registered through spark.plugins. Despite the docs showing spark.kubernetes.executor.resizeInterval as 0s, the config default in code is 1m.
• Every interval it reads each executor container's memory usage from the Kubernetes metrics API (metrics-server) and compares it with the container's memory limit.
• If usage > resizeThreshold * limit (default 0.9), it patches the pod's resize subresource, setting both the memory request and limit to limit * (1 + resizeFactor) (default 0.1).
• It only grows memory. There is no scale-down, and 4.2.0 has no cap. A resizeMaxMemory cap and support for the deployment pods allocator are in [SPARK-59918][K8S] Support deployment pods allocator in ExecutorResizePlugin spark#59180. In 4.2.0 the plugin only starts with the direct pods allocator.
• Spark's own pools are not resized. Heap and spark.memory.offHeap.size are fixed when the JVM starts, so the extra container memory is headroom for allocations that no Spark pool accounts for. That is exactly the memory Comet cannot size ahead of time.

Why this might help Comet

• A smaller baseline overhead, with the plugin as a safety net that grows it on demand, instead of sizing every executor for the worst case.
• Fewer exit-137 executor losses when Comet's untracked memory exceeds a static overhead.
• It needs no Comet code to try. Comet already documents memoryOverhead as the place this memory lives, so the first step is just measuring whether the combination works.

Open questions and likely gaps

1. Reaction time. The plugin polls every minute by default and grows 10% of the limit per step. Native memory in a sort, join, or shuffle can climb faster than that. How often does an executor die before the plugin reacts, and does a shorter interval or larger factor fix it?
2. What the plugin measures. It sees the container working set from metrics-server, which includes memory the allocator retains but could return, and it never shrinks. A long-running application could ratchet its limit up after one bad stage and keep it. Comet could reduce this by purging or decaying allocator pages between plans, which depends on the allocator work in Report resident memory (RssAnon) in the executor memory usage log and warn on it against the container size #6167 and Investigate adopting DataFusion's allocator-level memory accounting to replace manual memory tuning #4576.
3. A stale overhead in the warning. CometExecIterator computes the container's native budget from spark.executor.memoryOverhead (or the derived default) and warns when native memory exceeds it. After an in-place resize the real limit is larger than that, so the warning can fire on an executor that is fine. Reading the actual cgroup limit (memory.max on cgroup v2) instead of the conf would be correct with and without resizing.
4. A Comet-side trigger. The executor knows more than metrics-server does: RssAnon (proposed in Report resident memory (RssAnon) in the executor memory usage log and warn on it against the container size #6167), allocated, reserved, and the pool state. A Comet SparkPlugin could request a resize from the executor, either by patching its own pod (needs RBAC for pods/resize) or by sending the request to the driver over the plugin RPC channel and having the driver patch it. This would react faster and use a better signal than the generic plugin, at the cost of more code. It is worth building only if the experiment shows the generic plugin is too slow or too coarse.
5. Scale down. The Spark plugin never gives memory back. Lowering a memory limit in place is only safe once usage has dropped below the new limit, so a Comet-aware plugin would need to coordinate with allocator purging. This is probably out of scope for a first pass.
6. Prerequisites and failure modes. The cluster needs Kubernetes 1.33+ (1.35 for GA), metrics-server, and node capacity. A resize can be deferred or infeasible if the node is full, and the failure is only logged by the plugin. The docs should say what the user can expect when that happens.
7. Interaction with other memory work. This is complementary to Use Spark ResourceProfiles to give native and JVM stages independent memory configs #5116 (per-stage ResourceProfiles), which changes what is requested at stage boundaries, while resize changes it during a stage. It also overlaps with the circuit-breaker idea in Investigate adopting DataFusion's allocator-level memory accounting to replace manual memory tuning #4576, which is about acting on RSS inside the process.

Describe the potential solution

1. Experiment first. On a Kubernetes 1.35 cluster (kind or minikube is enough) with Spark 4.2.0 and a current Comet build, run a TPC-H or TPC-DS subset that is known to need a large untracked footprint, with spark.executor.memoryOverhead set well below what the memory usage log says is needed, in three configurations:

   • low overhead, no plugin
   • low overhead, spark.plugins=org.apache.spark.scheduler.cluster.k8s.ExecutorResizePlugin
   • correctly sized overhead, no plugin

   Record whether the query completes, the number of executor losses and exit codes, the resize events and time from threshold crossing to resize, final limit versus peak RssAnon, wall-clock time, and total memory-seconds. Repeat with a shorter resizeInterval and a larger resizeFactor.

2. Fix the stale limit in the native memory warning (open question 3), independent of the experiment's outcome.

3. Document it in the tuning guide and Kubernetes guide if the experiment shows it helps: when to use it, the recommended settings, the Kubernetes version and metrics-server prerequisites, and the direct allocator limit in 4.2.0.

4. Decide on a Comet-side plugin (open question 4) from the experiment results.

Additional context

• Spark: SPARK-55432 (ExecutorResizePlugin), SPARK-59918 (deployment allocator support), Spark 4.2.0 release notes, Kubernetes 1.35 in-place resize GA.
• Related Comet issues: Report resident memory (RssAnon) in the executor memory usage log and warn on it against the container size #6167 (report RssAnon in the memory usage log), Investigate adopting DataFusion's allocator-level memory accounting to replace manual memory tuning #4576 (allocator-level accounting and the RSS circuit breaker), Use Spark ResourceProfiles to give native and JVM stages independent memory configs #5116 (ResourceProfiles for native versus JVM stages), Kubernetes guide example never enables Comet because it sets no off-heap memory #6190 (Kubernetes guide example never enables Comet).