chore: deprecate spark.comet.exec.memoryPool.fraction

.github/workflows/pr_build_linux.yml

@@ -540,6 +540,7 @@ jobs:
               org.apache.spark.CometExecIteratorLifecycleSuite
               org.apache.spark.CometPluginsDefaultSuite
               org.apache.spark.CometPluginsMemoryOverheadWarningSuite
+              org.apache.spark.CometPluginsMemoryPoolFractionWarningSuite
               org.apache.spark.CometPluginsUnifiedModeSuite
               org.apache.comet.rules.CometScanRuleSuite
               org.apache.comet.rules.CometScanContribSuite

.github/workflows/pr_build_macos.yml

@@ -188,6 +188,7 @@ jobs:
               org.apache.spark.CometExecIteratorLifecycleSuite
               org.apache.spark.CometPluginsDefaultSuite
               org.apache.spark.CometPluginsMemoryOverheadWarningSuite
+              org.apache.spark.CometPluginsMemoryPoolFractionWarningSuite
               org.apache.spark.CometPluginsUnifiedModeSuite
               org.apache.comet.rules.CometScanRuleSuite
               org.apache.comet.rules.CometScanContribSuite

docs/source/contributor-guide/memory_management.md

@@ -198,10 +198,19 @@ rather than asking for a separate allocation:
 memory_limit = spark.memory.offHeap.size * spark.comet.exec.memoryPool.fraction
 ```
 
-`spark.comet.exec.memoryPool.fraction` defaults to `1.0`. Lowering it is the current workaround for
-Comet's under-accounting (see [The accounting gap](#the-accounting-gap)). It holds back a slice of
-the off-heap pool that Comet is not allowed to reserve, on the assumption that Comet's real usage
-overshoots its reservations by roughly that slice.
+`spark.comet.exec.memoryPool.fraction` defaults to `1.0` and is deprecated. It was documented as
+the workaround for Comet's under-accounting (see [The accounting gap](#the-accounting-gap)), holding
+back a slice of the off-heap pool for the memory Comet does not reserve, but it cannot do that:
+
+- `greedy_unified` ignores `memory_limit`. It asks Spark for every byte it reserves.
+- `fair_unified` applies it to each task's pool, limiting each memory consumer in the task to
+  `memory_limit / num_consumers`. Spark's execution pool already limits each of N running tasks to
+  `spark.memory.offHeap.size / N`, which is the tighter limit whenever more than one task is
+  running, and the tasks together can still acquire the whole pool.
+- Spark's own off-heap consumers, non-Comet operators and off-heap storage, draw on the same pool
+  with no Comet limit at all.
+
+The only room Spark leaves for memory outside the pool is `spark.executor.memoryOverhead`.
 
 A second value, `memory_limit_per_task`, is computed and passed alongside it, but only the on-heap
 pool types read it.
@@ -383,8 +392,9 @@ diverge for several structural reasons:
   outstanding.
 
 The practical consequence is that `reserved()` is a lower bound on Comet's real footprint, and the
-gap is workload-dependent. `spark.comet.exec.memoryPool.fraction` exists purely so operators can
-hand-tune a margin that covers the gap for their workload.
+gap is workload-dependent. The margin that covers it has to come from
+`spark.executor.memoryOverhead`. The deprecated `spark.comet.exec.memoryPool.fraction` cannot provide
+one; see [Where Comet's budget comes from](#where-comets-budget-comes-from).
 
 To measure the gap on a real workload, read the executor's periodic memory usage log, which
 reports the bytes Rust's allocator has handed out next to the pools' reservations; see
@@ -471,10 +481,11 @@ declared reservations are a lower bound on physical usage. The known gaps, rough
 much they matter:
 
 - **Real native usage is observed but not acted on.** Comet counts the bytes Rust's allocator has
-  handed out, and each executor logs that count next to the pools' reservations. Nothing reads it at runtime, though: no operator, metric, or policy
-  responds to it, so an executor that outgrows its container is still stopped only by the kill.
-- **`spark.comet.exec.memoryPool.fraction` is a manual proxy for the gap.** It asks operators to
-  guess a per-workload margin rather than measuring anything.
+  handed out, and each executor logs that count next to the pools' reservations. Nothing reads it
+  at runtime, though: no operator, metric, or policy responds to it, so an executor that outgrows
+  its container is still stopped only by the kill.
+- **The memory overhead is sized by hand.** The gap has to fit in `spark.executor.memoryOverhead`,
+  and the memory usage log measures it, but nothing sizes the overhead from it.
 - **`CometArrowAllocator` is unbounded** and participates in no budget.
 - **Buffer and reservation lifetimes are independent across the FFI boundary.** A batch can be
   resident on either side with no reservation covering it, because reservations are made and

docs/source/user-guide/latest/tuning.md

@@ -108,9 +108,13 @@ there is. That includes:
 
 Reserved memory is therefore a lower bound on what Comet really uses, and how far below it sits depends on the
 workload. This is why Comet can stay within the pool's limit and still push the executor past its container limit.
-Each executor logs how far apart the two are while Comet runs; see [Sizing the Overhead from the Memory Usage Log].
-To leave room for the part that is not counted, set `spark.comet.exec.memoryPool.fraction` to a value less than
-`1.0`, which restricts the amount of memory Comet is allowed to reserve.
+The part that is not counted has to fit in `spark.executor.memoryOverhead`, and each executor logs how large it is
+while Comet runs; see [Sizing the Overhead from the Memory Usage Log].
+
+`spark.comet.exec.memoryPool.fraction` is deprecated and does not leave room for it. Spark hands out all of
+`spark.memory.offHeap.size` to the tasks that ask for it, whatever the fraction. The `fair_unified` pool applies the
+fraction to each task separately, where Spark's own limit of an even share of the pool per running task is tighter
+whenever more than one task is running, and the `greedy_unified` pool ignores it.
 
 For more details about Spark off-heap memory mode, please refer to [Spark documentation].
 

spark/src/main/scala/org/apache/comet/CometConf.scala

@@ -862,9 +862,13 @@ object CometConf extends ShimCometConf {
     conf("spark.comet.exec.memoryPool.fraction")
       .category(CATEGORY_TUNING)
       .doc(
-        "Fraction of off-heap memory pool that is available to Comet. " +
-          "Only applies to off-heap mode. " +
-          s"$TUNING_GUIDE.")
+        "Deprecated: this config will be removed in a future release. It does not leave room " +
+          "in spark.memory.offHeap.size for native memory that Comet's memory pools do not " +
+          "track, because Spark hands out the whole off-heap pool whatever this is set to. Size " +
+          "spark.executor.memoryOverhead for that memory instead. Only applies to off-heap " +
+          "mode, where the fair_unified pool limits each memory consumer in a task to this " +
+          "fraction of the off-heap size divided by the task's consumers, and the " +
+          s"greedy_unified pool ignores it. $TUNING_GUIDE.")
       .doubleConf
       .createWithDefault(1.0)
 

spark/src/main/scala/org/apache/spark/Plugins.scala

@@ -73,6 +73,7 @@ class CometDriverPlugin extends DriverPlugin with Logging {
     CometDriverPlugin.registerCometMetrics(sc)
 
     CometDriverPlugin.warnIfExecutorMemoryOverheadUnset(sc.getConf)
+    CometDriverPlugin.warnIfMemoryPoolFractionSet(sc.getConf)
 
     extraConfs
   }
@@ -178,6 +179,22 @@ object CometDriverPlugin extends Logging {
     }
   }
 
+  // spark.comet.exec.memoryPool.fraction was documented as holding back part of the off-heap pool
+  // for the native memory that Comet does not reserve. It cannot: Spark hands out the whole pool
+  // to the tasks that ask for it, and the fraction only caps each task's consumers under
+  // fair_unified. Users who set it for that purpose need to size the memory overhead instead.
+  private[apache] def warnIfMemoryPoolFractionSet(conf: SparkConf): Unit = {
+    val key = CometConf.COMET_OFFHEAP_MEMORY_POOL_FRACTION.key
+    conf.getOption(key).foreach { value =>
+      logWarning(
+        s"$key=$value is deprecated and will be removed in a future release. It does not leave " +
+          "room in spark.memory.offHeap.size for native memory that Comet's memory pools do " +
+          "not track, because Spark hands out the whole off-heap pool whatever it is set to. " +
+          s"Size ${EXECUTOR_MEMORY_OVERHEAD.key} for that memory instead. " +
+          s"${CometConf.TUNING_GUIDE}.")
+    }
+  }
+
   private def getBooleanConf(conf: SparkConf, entry: ConfigEntry[Boolean]): Boolean =
     conf.getBoolean(entry.key, entry.defaultValue.get)
 

spark/src/test/scala/org/apache/spark/CometPluginsSuite.scala

@@ -202,6 +202,30 @@ class CometPluginsMemoryOverheadWarningSuite extends CometTestBase {
   }
 }
 
+class CometPluginsMemoryPoolFractionWarningSuite extends CometTestBase {
+
+  private val warning = "spark.comet.exec.memoryPool.fraction=0.8 is deprecated"
+
+  private def warningsFor(conf: SparkConf): Seq[String] = {
+    // Logging derives the logger name by stripping the object's trailing '$'
+    val logger = CometDriverPlugin.getClass.getName.stripSuffix("$")
+    val appender = new LogAppender("memory pool fraction warning")
+    withLogAppender(appender, Seq(logger), Some(Level.WARN)) {
+      CometDriverPlugin.warnIfMemoryPoolFractionSet(conf)
+    }
+    appender.loggingEvents.map(_.getMessage.getFormattedMessage).toSeq
+  }
+
+  test("warns when the memory pool fraction is set") {
+    val conf = new SparkConf().set("spark.comet.exec.memoryPool.fraction", "0.8")
+    assert(warningsFor(conf).exists(_.contains(warning)))
+  }
+
+  test("does not warn when the memory pool fraction is unset") {
+    assert(!warningsFor(new SparkConf()).exists(_.contains("memoryPool.fraction")))
+  }
+}
+
 class CometPluginsUnifiedModeSuite extends CometTestBase {
   override protected def sparkConf: SparkConf = {
     val conf = new SparkConf()