Consider accounting for Comet in spark.executor.memoryOverhead when off-heap memory is enabled

[andygrove]

What is the problem the feature request solves?

CometDriverPlugin adjusts spark.executor.memoryOverhead only when Comet runs in on-heap mode.
CometSparkSessionExtensions.shouldOverrideMemoryConf requires !offHeapMode, and
getCometMemoryOverheadInMiB returns 0 in off-heap mode, so the configuration we recommend to
everyone — off-heap memory enabled — gets no adjustment at all. The plugin logs "Comet is running in
unified memory mode and sharing off-heap memory with Spark" and leaves the overhead alone.

That is only half the picture for container sizing. Reservations Comet's operators make are charged
against spark.memory.offHeap.size, which the cluster manager already includes in the container
size, so those have room. Everything Comet allocates without reserving it does not: per-batch
working memory in expression kernels and Arrow array builders, decompression buffers and Parquet
reader structures, object store request buffers and the tokio runtime, Comet's JVM-side Arrow
buffers, and allocator overhead (padding, size-class rounding, fragmentation, retained pages). Those
come from the Rust global allocator and live in the native heap. Nothing in the container sizing
accounts for them, and spark.executor.memoryOverhead is the only slack the container has — slack
the JVM's own non-heap usage is already drawing on.

See What the container sees
in the memory management guide for the full accounting.

Describe the potential solution

Three options, in increasing order of risk:

1. Warn only. At driver init, when off-heap mode is enabled and spark.executor.memoryOverhead
   was not explicitly set, log a warning pointing at the tuning guide. No resource sizing changes.
   This mirrors CometDriverPlugin.warnIfKryoRegistratorMissing, which already handles a "set this
   before the context starts or you find out later" case in the same file.
2. Auto-adjust, opt-in. Extend the existing bump to off-heap mode behind a config that defaults
   to off.
3. Auto-adjust by default. Make getCometMemoryOverheadInMiB return a non-zero value in
   off-heap mode. This would silently grow every Comet user's container on Kubernetes and YARN,
   changing bin-packing, quotas, and capacity planning for clusters that are fine today.
   spark.comet.memoryOverhead is currently in CATEGORY_TESTING and documented as on-heap only,
   so it would need re-scoping and user-facing documentation as well.

Option 1 looks like the right starting point. The headroom actually needed is workload-dependent,
and overshooting it is a scheduling regression rather than a safe default.

Additional context

Worth verifying before building anything on options 2 or 3, because it may also mean the existing
on-heap bump does not do what the code intends: SparkContext materializes the default
ResourceProfile as the argument to
PluginContainer(this, _resourceProfileManager.defaultResourceProfile.resources), before any
plugin's init runs, and Kubernetes' BasicExecutorFeatureStep and YARN's allocator size executors
from the ResourceProfile rather than re-reading the conf. If that ordering holds on the Spark
versions Comet supports, the plugin's mutation updates sc.conf — which is all CometPluginsSuite
asserts, and it runs locally where nothing sizes a container — without changing the pod or container
spec.

The documentation side is handled separately: the tuning guide now has a "Configuring Executor
Memory Overhead" section and the spark-shell / spark-submit examples set
spark.executor.memoryOverhead=2g.

[andygrove]

Some history on how this came about, since it turns out no single change decided that off-heap mode needs no overhead.

#1379 introduced the off-heap distinction. In off-heap mode the plugin added getCometShuffleMemorySize rather than the full Comet overhead, and shouldOverrideMemoryConf gained && (!cometUnifiedMemoryManagerEnabled(conf) || !cometShuffleUnifiedMemoryManagerInTestEnabled(conf)), so off-heap mode still got something.

#1485 then removed the COMET_COLUMNAR_SHUFFLE_UNIFIED_MEMORY_ALLOCATOR_IN_TEST config and collapsed that predicate to a plain && !unifiedMemory. That is the point where off-heap mode stopped getting any overhead at all, and it happened as a side effect of deleting a test-only config rather than as a deliberate decision — the off-heap branch in the plugin became dead code.

#1525 deleted that dead branch and replaced it with assert(!isOffHeapEnabled(sc.getConf)), and #1561 produced the shape that is in main today: the early return 0 in getCometMemoryOverheadInMiB with the comment "when running in off-heap mode we use unified memory management to share off-heap memory with Spark so do not add overhead", plus the logInfo("Comet is running in unified memory mode and sharing off-heap memory with Spark") else-branch. #2599 later replaced the factor/minimum calculation with the flat spark.comet.memoryOverhead but left the off-heap return 0 alone.

So the comment in the code reads like a considered position, but it is really a description of where the refactoring landed. The premise behind it, that unified memory management means Comet needs no additional container headroom, only holds for the memory that operators actually reserve.

[andygrove]

Follow-up: #6053 is closed and #6054 replaces it.

Checking whether the adjustment reaches the executor container changed the answer. ResourceProfileManager materializes the default ResourceProfile in its constructor and snapshots spark.executor.memoryOverhead into it, and both the Kubernetes and YARN paths size executors from that profile rather than re-reading the conf. On Spark 3.4, 3.5 and 4.0 that happens before SparkContext builds the plugin container, so the existing on-heap adjustment has never reached the container on those versions. It only works on 4.1 and 4.2, where the two lines are ordered the other way round.

That makes option 3 from the description unattractive and option 1 the only one that works on every version, so #6054 deletes the adjustment and warns at startup when spark.executor.memoryOverhead is unset instead. It does change behavior on 4.1/4.2, where containers shrink by spark.comet.memoryOverhead unless the application sets the overhead itself.