This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6163-590b669a03937f748ee05c182d3f7d71142191a9 in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit ccc780a64095f1164b0cd4e71e57c7b49bb5aa17 Author: Andy Grove <[email protected]> AuthorDate: Thu Sep 24 10:08:45 2026 +0000 chore: deprecate spark.comet.exec.memoryPool.fraction (#6163) * feat: enable alloc-accounting by default and log executor native memory usage Build the alloc-accounting allocator wrapper by default (jemalloc stays opt-in) and have each executor periodically log its native memory usage while Comet native plans run: Comet native memory usage: allocated 5412.3 MiB, reserved 3890.0 MiB (16 native plans, 8 memory pools) The first CometExecIterator on an executor starts one daemon thread that logs a single line per interval for the whole executor, plus one more after the last plan finishes. It samples on a timer rather than between batches because a plan rooted at a native shuffle writer, or fed by native scans, can spend its whole run inside one executePlan call. The interval is spark.comet.memory.logInterval (default 10s, 0 disables), read from the executor's SparkConf. The new Native.getMemoryUsage JNI call reads only the allocation counter and the pool registry, and reads pool reservations after releasing the registry lock: CometFairMemoryPool holds its own lock across the JNI acquire from Spark, which can wait for a finishing task whose releasePlan needs the registry lock. Also add a tuning guide section on using the log to size spark.executor.memoryOverhead, update the tracing and memory management guides, and switch the rust-test CI step to cover the jemalloc arm and the build without the feature. * fix: always install allocation accounting, keep pool reads off the registry lock Remove the alloc-accounting cargo feature: the accounting allocator now wraps whichever backend the build selects, unconditionally. The alloc_overhead bench measures the wrapper against the bare backend in one binary, since there is no longer a build without it. releasePlan read the remaining pools' reservations while holding the pool registry lock on every plan, not only when tracing. With fair_unified that can deadlock: the pool holds its lock across a Spark acquire that waits on another task, whose releasePlan needs the registry lock. Every reader now copies the pools out and reads them after releasing it, and releasePlan reads nothing unless tracing. A malformed spark.comet.memory.logInterval now disables the log with a warning instead of failing every Comet task, and a value set only in the session is reported as ignored. The log also warns when the executor's native footprint (untracked native memory plus Spark's off-heap memory in use) exceeds spark.memory.offHeap.size plus the memory overhead. * docs: stop recommending memoryPool.fraction in the native memory warning The fraction does not reserve room in Spark's off-heap pool: greedy_unified ignores it, and fair_unified applies it per task, where Spark's own per-task limit is tighter. Point the warning and the tuning guide at the memory overhead instead. * chore: deprecate spark.comet.exec.memoryPool.fraction The fraction was documented as holding back part of the off-heap pool for native memory that Comet does not reserve, but it cannot: greedy_unified ignores it, fair_unified applies it per task where Spark's own per-task limit is tighter whenever more than one task runs, and Spark hands out the whole pool either way. The only room Spark leaves for untracked native memory is spark.executor.memoryOverhead. Mark the setting deprecated, keep its current behaviour, have the driver plugin warn when it is set, and correct the tuning and memory management guides. --- .github/workflows/pr_build_linux.yml | 1 + .github/workflows/pr_build_macos.yml | 1 + docs/source/contributor-guide/memory_management.md | 31 +++++++++++++++------- docs/source/user-guide/latest/tuning.md | 10 ++++--- .../main/scala/org/apache/comet/CometConf.scala | 10 ++++--- .../src/main/scala/org/apache/spark/Plugins.scala | 17 ++++++++++++ .../scala/org/apache/spark/CometPluginsSuite.scala | 24 +++++++++++++++++ 7 files changed, 78 insertions(+), 16 deletions(-) diff --git a/.github/workflows/pr_build_linux.yml b/.github/workflows/pr_build_linux.yml index da9a940d9d..8abd6c9249 100644 --- a/.github/workflows/pr_build_linux.yml +++ b/.github/workflows/pr_build_linux.yml @@ -541,6 +541,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 diff --git a/.github/workflows/pr_build_macos.yml b/.github/workflows/pr_build_macos.yml index d74491f09a..af6bbc48c8 100644 --- a/.github/workflows/pr_build_macos.yml +++ b/.github/workflows/pr_build_macos.yml @@ -189,6 +189,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 diff --git a/docs/source/contributor-guide/memory_management.md b/docs/source/contributor-guide/memory_management.md index 93ba4e8754..daa7a0f0f6 100644 --- a/docs/source/contributor-guide/memory_management.md +++ b/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 diff --git a/docs/source/user-guide/latest/tuning.md b/docs/source/user-guide/latest/tuning.md index 5855d6cdc0..2ebe8987c5 100644 --- a/docs/source/user-guide/latest/tuning.md +++ b/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]. diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala b/spark/src/main/scala/org/apache/comet/CometConf.scala index fa77058154..ebc3b068cb 100644 --- a/spark/src/main/scala/org/apache/comet/CometConf.scala +++ b/spark/src/main/scala/org/apache/comet/CometConf.scala @@ -902,9 +902,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) diff --git a/spark/src/main/scala/org/apache/spark/Plugins.scala b/spark/src/main/scala/org/apache/spark/Plugins.scala index 24b6ab3eea..d040208596 100644 --- a/spark/src/main/scala/org/apache/spark/Plugins.scala +++ b/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) diff --git a/spark/src/test/scala/org/apache/spark/CometPluginsSuite.scala b/spark/src/test/scala/org/apache/spark/CometPluginsSuite.scala index 278c281b1d..4673740daf 100644 --- a/spark/src/test/scala/org/apache/spark/CometPluginsSuite.scala +++ b/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() --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
