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-6054-d62d3382ac5db67f185dc9f77544a5cefd2002ab in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 67c23a219451d76496c9be19f3fb61b280a1f51e Author: Andy Grove <[email protected]> AuthorDate: Sun Sep 20 15:23:47 2026 +0000 fix: remove the ineffective spark.executor.memoryOverhead adjustment from the driver plugin (#6054) * fix: remove the ineffective spark.executor.memoryOverhead adjustment from the driver plugin * fix: drop a redundant string interpolator --- .github/workflows/pr_build_linux.yml | 4 +- .github/workflows/pr_build_macos.yml | 4 +- .../main/scala/org/apache/comet/CometConf.scala | 2 +- .../apache/comet/CometSparkSessionExtensions.scala | 20 +---- .../src/main/scala/org/apache/spark/Plugins.scala | 67 ++++++++-------- .../spark/comet/shims/ShimCometDriverPlugin.scala | 31 -------- .../spark/comet/shims/ShimCometDriverPlugin.scala | 28 ------- .../comet/CometSparkSessionExtensionsSuite.scala | 3 +- .../scala/org/apache/spark/CometPluginsSuite.scala | 89 +++++++++++----------- 9 files changed, 87 insertions(+), 161 deletions(-) diff --git a/.github/workflows/pr_build_linux.yml b/.github/workflows/pr_build_linux.yml index cbb73fab98..17c9da123b 100644 --- a/.github/workflows/pr_build_linux.yml +++ b/.github/workflows/pr_build_linux.yml @@ -539,8 +539,8 @@ jobs: org.apache.spark.CometTaskMemoryManagerSuite org.apache.spark.CometExecIteratorLifecycleSuite org.apache.spark.CometPluginsDefaultSuite - org.apache.spark.CometPluginsNonOverrideSuite - org.apache.spark.CometPluginsUnifiedModeOverrideSuite + org.apache.spark.CometPluginsMemoryOverheadWarningSuite + org.apache.spark.CometPluginsUnifiedModeSuite org.apache.comet.rules.CometScanRuleSuite org.apache.comet.rules.CometScanContribSuite org.apache.comet.rules.CometScanSchemeFallbackSuite diff --git a/.github/workflows/pr_build_macos.yml b/.github/workflows/pr_build_macos.yml index bcf73b8a56..5a1abca8cf 100644 --- a/.github/workflows/pr_build_macos.yml +++ b/.github/workflows/pr_build_macos.yml @@ -187,8 +187,8 @@ jobs: org.apache.spark.CometTaskMemoryManagerSuite org.apache.spark.CometExecIteratorLifecycleSuite org.apache.spark.CometPluginsDefaultSuite - org.apache.spark.CometPluginsNonOverrideSuite - org.apache.spark.CometPluginsUnifiedModeOverrideSuite + org.apache.spark.CometPluginsMemoryOverheadWarningSuite + org.apache.spark.CometPluginsUnifiedModeSuite org.apache.comet.rules.CometScanRuleSuite org.apache.comet.rules.CometScanContribSuite org.apache.comet.rules.CometScanSchemeFallbackSuite diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala b/spark/src/main/scala/org/apache/comet/CometConf.scala index d2369a13c2..d371f1ba44 100644 --- a/spark/src/main/scala/org/apache/comet/CometConf.scala +++ b/spark/src/main/scala/org/apache/comet/CometConf.scala @@ -48,7 +48,7 @@ object CometConf extends ShimCometConf { val COMPAT_GUIDE: String = "For more information, refer to the Comet Compatibility " + "Guide (https://datafusion.apache.org/comet/user-guide/latest/compatibility/index.html)" - private val TUNING_GUIDE = "For more information, refer to the Comet Tuning " + + val TUNING_GUIDE: String = "For more information, refer to the Comet Tuning " + "Guide (https://datafusion.apache.org/comet/user-guide/latest/tuning.html)" private val TRACING_GUIDE = "For more information, refer to the Comet Tracing " + diff --git a/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala b/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala index 5a67d5eef8..4b37d7b61a 100644 --- a/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala +++ b/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala @@ -247,27 +247,14 @@ object CometSparkSessionExtensions extends Logging { org.apache.spark.SPARK_VERSION >= "4.2" } - /** - * Whether we should override Spark memory configuration for Comet. This only returns true when - * Comet native execution is enabled and/or Comet shuffle is enabled and Comet doesn't use - * off-heap mode (unified memory manager). - */ - def shouldOverrideMemoryConf(conf: SparkConf): Boolean = { - val cometEnabled = getBooleanConf(conf, CometConf.COMET_ENABLED) - val cometShuffleEnabled = getBooleanConf(conf, CometConf.COMET_SHUFFLE_ENABLED) - val cometExecEnabled = getBooleanConf(conf, CometConf.COMET_EXEC_ENABLED) - val offHeapMode = CometSparkSessionExtensions.isOffHeapEnabled(conf) - cometEnabled && (cometShuffleEnabled || cometExecEnabled) && !offHeapMode - } - /** * Determines required memory overhead in MB per executor process for Comet when running in * on-heap mode. */ def getCometMemoryOverheadInMiB(sparkConf: SparkConf): Long = { if (isOffHeapEnabled(sparkConf)) { - // when running in off-heap mode we use unified memory management to share - // off-heap memory with Spark so do not add overhead + // off-heap mode sizes the native memory pool from spark.memory.offHeap.size instead + // (see CometExecIterator.getMemoryConfig), so this value does not apply return 0 } ConfigHelpers.byteFromString( @@ -277,9 +264,6 @@ object CometSparkSessionExtensions extends Logging { ByteUnit.MiB) } - private def getBooleanConf(conf: SparkConf, entry: ConfigEntry[Boolean]) = - conf.getBoolean(entry.key, entry.defaultValue.get) - /** * Calculates required memory overhead in bytes per executor process for Comet when running in * on-heap mode. diff --git a/spark/src/main/scala/org/apache/spark/Plugins.scala b/spark/src/main/scala/org/apache/spark/Plugins.scala index eaeac31665..24b6ab3eea 100644 --- a/spark/src/main/scala/org/apache/spark/Plugins.scala +++ b/spark/src/main/scala/org/apache/spark/Plugins.scala @@ -23,13 +23,12 @@ import java.{util => ju} import java.util.Collections import org.apache.spark.api.plugin.{DriverPlugin, ExecutorPlugin, PluginContext, SparkPlugin} -import org.apache.spark.comet.shims.ShimCometDriverPlugin import org.apache.spark.internal.Logging -import org.apache.spark.internal.config.{EXECUTOR_MEMORY, EXECUTOR_MEMORY_OVERHEAD, EXECUTOR_MEMORY_OVERHEAD_FACTOR} +import org.apache.spark.internal.config.EXECUTOR_MEMORY_OVERHEAD import org.apache.spark.sql.internal.StaticSQLConf import org.apache.comet.{COMET_VERSION, CometSparkSessionExtensions, NativeBase} -import org.apache.comet.CometConf +import org.apache.comet.{CometConf, ConfigEntry} import org.apache.comet.CometConf.{COMET_METRICS_ENABLED, COMET_ONHEAP_ENABLED} import org.apache.comet.CometKryoRegistrator import org.apache.comet.annotation.Public @@ -45,8 +44,7 @@ import org.apache.comet.annotation.Public * * To enable this plugin, set the config "spark.plugins" to `org.apache.spark.CometPlugin`. */ -class CometDriverPlugin extends DriverPlugin with Logging with ShimCometDriverPlugin { - private val EXECUTOR_MEMORY_DEFAULT = "1g" +class CometDriverPlugin extends DriverPlugin with Logging { override def init(sc: SparkContext, pluginContext: PluginContext): ju.Map[String, String] = { logInfo("CometDriverPlugin init") @@ -74,32 +72,7 @@ class CometDriverPlugin extends DriverPlugin with Logging with ShimCometDriverPl // Register Comet metrics CometDriverPlugin.registerCometMetrics(sc) - if (CometSparkSessionExtensions.shouldOverrideMemoryConf(sc.getConf)) { - val execMemOverhead = if (sc.getConf.contains(EXECUTOR_MEMORY_OVERHEAD.key)) { - sc.getConf.getSizeAsMb(EXECUTOR_MEMORY_OVERHEAD.key) - } else { - // By default, executorMemory * spark.executor.memoryOverheadFactor, with minimum of 384MB - val executorMemory = - sc.getConf.getSizeAsMb(EXECUTOR_MEMORY.key, EXECUTOR_MEMORY_DEFAULT) - val memoryOverheadFactor = sc.getConf.get(EXECUTOR_MEMORY_OVERHEAD_FACTOR) - val memoryOverheadMinMib = getMemoryOverheadMinMib(sc.getConf) - - Math.max((executorMemory * memoryOverheadFactor).toLong, memoryOverheadMinMib) - } - - val cometMemOverhead = CometSparkSessionExtensions.getCometMemoryOverheadInMiB(sc.getConf) - sc.conf.set(EXECUTOR_MEMORY_OVERHEAD.key, s"${execMemOverhead + cometMemOverhead}M") - val newExecMemOverhead = sc.getConf.getSizeAsMb(EXECUTOR_MEMORY_OVERHEAD.key) - - logInfo(s""" - Overriding Spark memory configuration for Comet: - - Spark executor memory overhead: ${execMemOverhead}MB - - Comet memory overhead: ${cometMemOverhead}MB - - Updated Spark executor memory overhead: ${newExecMemOverhead}MB - """) - } else { - logInfo("Comet is running in unified memory mode and sharing off-heap memory with Spark") - } + CometDriverPlugin.warnIfExecutorMemoryOverheadUnset(sc.getConf) extraConfs } @@ -176,6 +149,38 @@ object CometDriverPlugin extends Logging { } } + // Comet's native allocations are made by the Rust global allocator and live in the native heap. + // The share that operators reserve is charged against a memory pool, but everything else -- + // expression kernels and Arrow array builders, decompression buffers, Parquet reader structures, + // object store buffers, the tokio runtime, allocator overhead -- is covered by no budget at all, + // and neither is Comet's JVM-side Arrow allocator. The only slack the executor container has for + // that is spark.executor.memoryOverhead, which the JVM's own non-heap usage already draws on. + // + // Comet used to add spark.comet.memoryOverhead to it here, but a driver plugin cannot: on Spark + // 3.4, 3.5 and 4.0, SparkContext builds the default ResourceProfile before it creates the plugin + // container, and the cluster managers size executors from that profile rather than re-reading + // the conf, so the new value never reached the container. Say so while the application is still + // starting up instead, because this has to be set before the SparkContext is created. + private[apache] def warnIfExecutorMemoryOverheadUnset(conf: SparkConf): Unit = { + val cometEnabled = getBooleanConf(conf, CometConf.COMET_ENABLED) + val cometExecEnabled = getBooleanConf(conf, CometConf.COMET_EXEC_ENABLED) + val cometShuffleEnabled = getBooleanConf(conf, CometConf.COMET_SHUFFLE_ENABLED) + val cometActive = cometEnabled && (cometExecEnabled || cometShuffleEnabled) + + if (cometActive && !conf.contains(EXECUTOR_MEMORY_OVERHEAD.key)) { + logWarning( + s"${EXECUTOR_MEMORY_OVERHEAD.key} is not set. Comet allocates outside the JVM heap, and " + + "the part of that which no memory pool tracks is not covered by " + + "spark.executor.memory or spark.memory.offHeap.size, so Spark's default overhead can " + + "leave the executor short and the cluster manager may kill it. Set " + + s"${EXECUTOR_MEMORY_OVERHEAD.key} before creating the SparkContext; it cannot be set " + + s"later. ${CometConf.TUNING_GUIDE}.") + } + } + + private def getBooleanConf(conf: SparkConf, entry: ConfigEntry[Boolean]): Boolean = + conf.getBoolean(entry.key, entry.defaultValue.get) + def registerCometMetrics(sc: SparkContext): Unit = { if (sc.getConf.getBoolean( COMET_METRICS_ENABLED.key, diff --git a/spark/src/main/spark-3.x/org/apache/spark/comet/shims/ShimCometDriverPlugin.scala b/spark/src/main/spark-3.x/org/apache/spark/comet/shims/ShimCometDriverPlugin.scala deleted file mode 100644 index d6a7e2d7f4..0000000000 --- a/spark/src/main/spark-3.x/org/apache/spark/comet/shims/ShimCometDriverPlugin.scala +++ /dev/null @@ -1,31 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.spark.comet.shims - -import org.apache.spark.SparkConf - -trait ShimCometDriverPlugin { - // `org.apache.spark.internal.config.EXECUTOR_MIN_MEMORY_OVERHEAD` was added since Spark 4.0.0 - private val EXECUTOR_MIN_MEMORY_OVERHEAD = "spark.executor.minMemoryOverhead" - private val EXECUTOR_MIN_MEMORY_OVERHEAD_DEFAULT = 384L - - def getMemoryOverheadMinMib(sc: SparkConf): Long = - sc.getLong(EXECUTOR_MIN_MEMORY_OVERHEAD, EXECUTOR_MIN_MEMORY_OVERHEAD_DEFAULT) -} diff --git a/spark/src/main/spark-4.x/org/apache/spark/comet/shims/ShimCometDriverPlugin.scala b/spark/src/main/spark-4.x/org/apache/spark/comet/shims/ShimCometDriverPlugin.scala deleted file mode 100644 index 4e48744fc4..0000000000 --- a/spark/src/main/spark-4.x/org/apache/spark/comet/shims/ShimCometDriverPlugin.scala +++ /dev/null @@ -1,28 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.spark.comet.shims - -import org.apache.spark.SparkConf -import org.apache.spark.internal.config.EXECUTOR_MIN_MEMORY_OVERHEAD - -trait ShimCometDriverPlugin { - protected def getMemoryOverheadMinMib(sparkConf: SparkConf): Long = - sparkConf.get(EXECUTOR_MIN_MEMORY_OVERHEAD) -} diff --git a/spark/src/test/scala/org/apache/comet/CometSparkSessionExtensionsSuite.scala b/spark/src/test/scala/org/apache/comet/CometSparkSessionExtensionsSuite.scala index 0c9c1e1706..331d4b4ece 100644 --- a/spark/src/test/scala/org/apache/comet/CometSparkSessionExtensionsSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometSparkSessionExtensionsSuite.scala @@ -166,7 +166,6 @@ class CometSparkSessionExtensionsSuite extends CometTestBase { val sparkConf = new SparkConf() sparkConf.set(CometConf.COMET_ONHEAP_MEMORY_OVERHEAD.key, "10g") assert(getCometMemoryOverhead(sparkConf) == getBytesFromMib(1024 * 10)) - assert(shouldOverrideMemoryConf(sparkConf)) } test("Comet memory overhead (off heap)") { @@ -174,8 +173,8 @@ class CometSparkSessionExtensionsSuite extends CometTestBase { sparkConf.set(CometConf.COMET_ONHEAP_MEMORY_OVERHEAD.key, "64g") sparkConf.set("spark.memory.offHeap.enabled", "true") sparkConf.set("spark.memory.offHeap.size", "10g") + // off-heap mode sizes the native pool from spark.memory.offHeap.size instead assert(getCometMemoryOverhead(sparkConf) == 0) - assert(!shouldOverrideMemoryConf(sparkConf)) } test("Comet shuffle memory factor") { diff --git a/spark/src/test/scala/org/apache/spark/CometPluginsSuite.scala b/spark/src/test/scala/org/apache/spark/CometPluginsSuite.scala index 9a97c35067..278c281b1d 100644 --- a/spark/src/test/scala/org/apache/spark/CometPluginsSuite.scala +++ b/spark/src/test/scala/org/apache/spark/CometPluginsSuite.scala @@ -21,6 +21,7 @@ package org.apache.spark import java.io.File +import org.apache.logging.log4j.Level import org.apache.spark.sql.{CometTestBase, SaveMode} import org.apache.spark.sql.internal.StaticSQLConf @@ -129,17 +130,19 @@ class CometPluginsSuite extends CometTestBase { } } - test("Default Comet memory overhead") { + test("executor memory overhead is left alone") { + // Comet does not adjust spark.executor.memoryOverhead. A driver plugin runs too late to + // influence the executor container on Spark 3.4, 3.5 and 4.0, so the value the application + // set is the value that is used. val execMemOverhead1 = spark.conf.get("spark.executor.memoryOverhead") val execMemOverhead2 = spark.sessionState.conf.getConfString("spark.executor.memoryOverhead") val execMemOverhead3 = spark.sparkContext.getConf.get("spark.executor.memoryOverhead") val execMemOverhead4 = spark.sparkContext.conf.get("spark.executor.memoryOverhead") - // 2GB + 384MB (default Comet memory overhead) - assert(execMemOverhead1 == "3072M") - assert(execMemOverhead2 == "3072M") - assert(execMemOverhead3 == "3072M") - assert(execMemOverhead4 == "3072M") + assert(execMemOverhead1 == "2G") + assert(execMemOverhead2 == "2G") + assert(execMemOverhead3 == "2G") + assert(execMemOverhead4 == "2G") } } @@ -156,50 +159,50 @@ class CometPluginsDefaultSuite extends CometTestBase { conf } - test("Default executor memory overhead + Comet memory overhead") { - val execMemOverhead1 = spark.conf.get("spark.executor.memoryOverhead") - val execMemOverhead2 = spark.sessionState.conf.getConfString("spark.executor.memoryOverhead") - val execMemOverhead3 = spark.sparkContext.getConf.get("spark.executor.memoryOverhead") - val execMemOverhead4 = spark.sparkContext.conf.get("spark.executor.memoryOverhead") - - // Spark executor memory overhead = executor memory (1G) * memoryOverheadFactor (0.5) = 512MB - // 512MB + 384MB (default Comet memory overhead) - assert(execMemOverhead1 == "1536M") - assert(execMemOverhead2 == "1536M") - assert(execMemOverhead3 == "1536M") - assert(execMemOverhead4 == "1536M") + test("unset executor memory overhead is left unset") { + assert(!spark.sparkContext.conf.contains("spark.executor.memoryOverhead")) } } -class CometPluginsNonOverrideSuite extends CometTestBase { - override protected def sparkConf: SparkConf = { +class CometPluginsMemoryOverheadWarningSuite extends CometTestBase { + + private val warning = "spark.executor.memoryOverhead is not set" + + 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("executor memory overhead warning") + withLogAppender(appender, Seq(logger), Some(Level.WARN)) { + CometDriverPlugin.warnIfExecutorMemoryOverheadUnset(conf) + } + appender.loggingEvents.map(_.getMessage.getFormattedMessage).toSeq + } + + test("warns when executor memory overhead is unset and Comet is active") { val conf = new SparkConf() - conf.set("spark.driver.memory", "1G") - conf.set("spark.executor.memory", "1G") - conf.set("spark.executor.memoryOverhead", "2G") - conf.set("spark.executor.memoryOverheadFactor", "0.5") - conf.set("spark.plugins", "org.apache.spark.CometPlugin") conf.set("spark.comet.enabled", "true") - conf.set("spark.comet.shuffle.enabled", "false") - conf.set("spark.comet.exec.enabled", "false") - conf.set("spark.comet.exec.onHeap.enabled", "true") - conf + conf.set("spark.comet.exec.enabled", "true") + assert(warningsFor(conf).exists(_.contains(warning))) } - test("executor memory overhead is not overridden") { - val execMemOverhead1 = spark.conf.get("spark.executor.memoryOverhead") - val execMemOverhead2 = spark.sessionState.conf.getConfString("spark.executor.memoryOverhead") - val execMemOverhead3 = spark.sparkContext.getConf.get("spark.executor.memoryOverhead") - val execMemOverhead4 = spark.sparkContext.conf.get("spark.executor.memoryOverhead") + test("does not warn when executor memory overhead is set") { + val conf = new SparkConf() + conf.set("spark.comet.enabled", "true") + conf.set("spark.comet.exec.enabled", "true") + conf.set("spark.executor.memoryOverhead", "2g") + assert(!warningsFor(conf).exists(_.contains(warning))) + } - assert(execMemOverhead1 == "2G") - assert(execMemOverhead2 == "2G") - assert(execMemOverhead3 == "2G") - assert(execMemOverhead4 == "2G") + test("does not warn when Comet is not executing anything") { + val conf = new SparkConf() + conf.set("spark.comet.enabled", "true") + conf.set("spark.comet.exec.enabled", "false") + conf.set("spark.comet.shuffle.enabled", "false") + assert(!warningsFor(conf).exists(_.contains(warning))) } } -class CometPluginsUnifiedModeOverrideSuite extends CometTestBase { +class CometPluginsUnifiedModeSuite extends CometTestBase { override protected def sparkConf: SparkConf = { val conf = new SparkConf() conf.set("spark.driver.memory", "1G") @@ -211,21 +214,15 @@ class CometPluginsUnifiedModeOverrideSuite extends CometTestBase { conf.set("spark.memory.offHeap.size", "2G") conf.set("spark.comet.shuffle.enabled", "true") conf.set("spark.comet.exec.enabled", "true") - conf.set("spark.comet.memory.overhead.factor", "0.5") conf } - /* - * Since using unified memory executor memory should not be overridden - */ - test("executor memory overhead is not overridden") { + test("executor memory overhead is left alone in off-heap mode") { val execMemOverhead1 = spark.conf.get("spark.executor.memoryOverhead") val execMemOverhead2 = spark.sessionState.conf.getConfString("spark.executor.memoryOverhead") val execMemOverhead3 = spark.sparkContext.getConf.get("spark.executor.memoryOverhead") val execMemOverhead4 = spark.sparkContext.conf.get("spark.executor.memoryOverhead") - // in unified memory mode, comet memory overhead is - // spark.memory.offHeap.size (2G) * spark.comet.memory.overhead.factor (0.5) = 1G and the overhead is not overridden assert(execMemOverhead1 == "1G") assert(execMemOverhead2 == "1G") assert(execMemOverhead3 == "1G") --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
