This is an automated email from the ASF dual-hosted git repository.
jackylee-ch pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 44c6048252 [GLUTEN-12676][CORE] Read MiB-declared Spark memory configs
with matching accessors (#12677)
44c6048252 is described below
commit 44c6048252a95239027a48dbfe5491b603cae8c0
Author: YangJie <[email protected]>
AuthorDate: Thu Aug 6 10:20:31 2026 +0800
[GLUTEN-12676][CORE] Read MiB-declared Spark memory configs with matching
accessors (#12677)
---
.../scala/org/apache/gluten/GlutenPlugin.scala | 8 +-
.../apache/gluten/config/GlutenCoreConfig.scala | 2 -
.../org/apache/spark/util/SparkResourceUtil.scala | 35 +++++-
.../gluten/GlutenDynamicOffHeapSizingSuite.scala | 55 +++++++++
.../apache/spark/util/SparkResourceUtilSuite.scala | 85 ++++++++++++++
.../GlutenAutoAdjustStageResourceProfile.scala | 50 ++++++--
...GlutenAutoAdjustStageResourceProfileSuite.scala | 129 +++++++++++++++++++++
7 files changed, 341 insertions(+), 23 deletions(-)
diff --git a/gluten-core/src/main/scala/org/apache/gluten/GlutenPlugin.scala
b/gluten-core/src/main/scala/org/apache/gluten/GlutenPlugin.scala
index 9cc283bd25..81fea4727b 100644
--- a/gluten-core/src/main/scala/org/apache/gluten/GlutenPlugin.scala
+++ b/gluten-core/src/main/scala/org/apache/gluten/GlutenPlugin.scala
@@ -111,13 +111,7 @@ private object GlutenDriverPlugin extends Logging {
// Get the off-heap size set by user.
val offHeapSize =
if (conf.getBoolean(GlutenCoreConfig.DYNAMIC_OFFHEAP_SIZING_ENABLED.key,
false)) {
- val onHeapSize: Long =
- if (conf.contains(GlutenCoreConfig.SPARK_ONHEAP_SIZE_KEY)) {
- conf.getSizeAsBytes(GlutenCoreConfig.SPARK_ONHEAP_SIZE_KEY)
- } else {
- // 1GB default
- 1024 * 1024 * 1024
- }
+ val onHeapSize: Long = SparkResourceUtil.getExecutorMemorySize(conf)
if (conf.contains(GlutenCoreConfig.SPARK_OFFHEAP_ENABLED_KEY)) {
logWarning(
diff --git
a/gluten-core/src/main/scala/org/apache/gluten/config/GlutenCoreConfig.scala
b/gluten-core/src/main/scala/org/apache/gluten/config/GlutenCoreConfig.scala
index eb08396772..ff96b5e4b6 100644
--- a/gluten-core/src/main/scala/org/apache/gluten/config/GlutenCoreConfig.scala
+++ b/gluten-core/src/main/scala/org/apache/gluten/config/GlutenCoreConfig.scala
@@ -76,8 +76,6 @@ object GlutenCoreConfig extends ConfigRegistry {
val SPARK_OFFHEAP_SIZE_KEY = "spark.memory.offHeap.size"
val SPARK_OFFHEAP_ENABLED_KEY = "spark.memory.offHeap.enabled"
- val SPARK_ONHEAP_SIZE_KEY = "spark.executor.memory"
-
val GLUTEN_ENABLED =
buildConf("spark.gluten.enabled")
.doc(
diff --git
a/gluten-core/src/main/scala/org/apache/spark/util/SparkResourceUtil.scala
b/gluten-core/src/main/scala/org/apache/spark/util/SparkResourceUtil.scala
index b25a174c12..c27000beca 100644
--- a/gluten-core/src/main/scala/org/apache/spark/util/SparkResourceUtil.scala
+++ b/gluten-core/src/main/scala/org/apache/spark/util/SparkResourceUtil.scala
@@ -105,9 +105,40 @@ object SparkResourceUtil extends Logging {
val executorMemMib = conf.get(EXECUTOR_MEMORY)
val factor =
conf.getDouble(MEMORY_OVERHEAD_FACTOR, 0.1d)
- val minMib = conf.getLong(MIN_MEMORY_OVERHEAD, 384L)
+ // spark.executor.minMemoryOverhead is a size string, MiB unless
suffixed, so getLong would
+ // fail to parse anything carrying a unit. getSizeAsMb keeps the value
in MiB, which is what
+ // the max below compares against.
+ val minMib = conf.getSizeAsMb(MIN_MEMORY_OVERHEAD, "384m")
(executorMemMib * factor).toLong.max(minMib)
}
- ByteUnit.MiB.toBytes(overheadMib)
+ mibToBytes(overheadMib)
+ }
+
+ /**
+ * Returns spark.executor.memory in bytes.
+ *
+ * Spark declares the config as bytesConf(ByteUnit.MiB), so a value without
a size suffix means
+ * MiB, and SparkConf#getSizeAsBytes must not be used to read it. This
matches how Spark itself
+ * reads the config on YARN and K8s. Standalone and local-cluster differ:
there Spark resolves it
+ * through the executor-memory-in-MiB path (SparkContext#executorMemoryInMb
on Spark 3.4+), which
+ * treats a suffix-less value as bytes.
+ */
+ def getExecutorMemorySize(conf: SparkConf): Long = {
+ val memoryMib = conf.get(EXECUTOR_MEMORY)
+ require(memoryMib >= 0, s"${EXECUTOR_MEMORY.key} should not be negative,
but was $memoryMib")
+ mibToBytes(memoryMib)
+ }
+
+ /**
+ * Converts a MiB-valued Spark memory amount to bytes.
+ *
+ * ByteUnit#toBytes rejects a negative input but wraps silently on overflow,
while
+ * ByteUnit#convertTo raises on overflow but passes a negative through, so
guard the sign here and
+ * let convertTo guard the magnitude. Callers that can name the offending
config add their own
+ * require first so that the more specific message wins.
+ */
+ def mibToBytes(memoryMib: Long): Long = {
+ require(memoryMib >= 0, s"Memory size in MiB should not be negative, but
was $memoryMib")
+ ByteUnit.MiB.convertTo(memoryMib, ByteUnit.BYTE)
}
}
diff --git
a/gluten-core/src/test/scala/org/apache/gluten/GlutenDynamicOffHeapSizingSuite.scala
b/gluten-core/src/test/scala/org/apache/gluten/GlutenDynamicOffHeapSizingSuite.scala
new file mode 100644
index 0000000000..6095eca27b
--- /dev/null
+++
b/gluten-core/src/test/scala/org/apache/gluten/GlutenDynamicOffHeapSizingSuite.scala
@@ -0,0 +1,55 @@
+/*
+ * 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.gluten
+
+import org.apache.gluten.component.WithDummyBackend
+import org.apache.gluten.config.GlutenCoreConfig
+
+import org.apache.spark.{SparkConf, SparkContext}
+
+import org.scalatest.funsuite.AnyFunSuite
+
+class GlutenDynamicOffHeapSizingSuite extends AnyFunSuite with
WithDummyBackend {
+
+ private val MIB = 1024L * 1024L
+
+ test("dynamic off-heap sizing derives the budget from a suffix-less executor
memory") {
+ // spark.executor.memory is MiB-unless-suffixed. Reading 8192 as bytes
made the budget
+ // negative, so drive the real driver-init path and assert the derived
budget. spark.testing
+ // zeroes Spark's reserved memory; without it UnifiedMemoryManager rejects
8192 bytes before
+ // the plugin runs, which is the very read this test pins down.
+ val conf = new SparkConf(false)
+ .setAppName("GlutenDynamicOffHeapSizingSuite")
+ .set("spark.master", "local[1]")
+ .set("spark.plugins", classOf[GlutenPlugin].getName)
+ .set("spark.testing", "true")
+ .set("spark.ui.enabled", "false")
+ .set(GlutenCoreConfig.DYNAMIC_OFFHEAP_SIZING_ENABLED.key, "true")
+ .set("spark.executor.memory", "8192")
+ val sc = new SparkContext(conf)
+ try {
+ val expected = ((8192L * MIB - 300L * MIB) * 0.6d).toLong
+ assert(expected == 4965217075L)
+ assert(
+
sc.getConf.getLong(GlutenCoreConfig.COLUMNAR_OFFHEAP_SIZE_IN_BYTES.key, -1L) ==
expected)
+ assert(
+
sc.getConf.getLong(GlutenCoreConfig.COLUMNAR_TASK_OFFHEAP_SIZE_IN_BYTES.key,
-1L) > 0L)
+ } finally {
+ sc.stop()
+ }
+ }
+}
diff --git
a/gluten-core/src/test/scala/org/apache/spark/util/SparkResourceUtilSuite.scala
b/gluten-core/src/test/scala/org/apache/spark/util/SparkResourceUtilSuite.scala
index bad2153ddc..48d10abe9a 100644
---
a/gluten-core/src/test/scala/org/apache/spark/util/SparkResourceUtilSuite.scala
+++
b/gluten-core/src/test/scala/org/apache/spark/util/SparkResourceUtilSuite.scala
@@ -65,4 +65,89 @@ class SparkResourceUtilSuite extends AnyFunSuite {
val conf = new SparkConf(false).set("spark.master", "local[8]")
assert(SparkResourceUtil.getTaskSlots(conf) == 8)
}
+
+ test("getExecutorMemorySize reads a bare spark.executor.memory as MiB") {
+ // Spark defines spark.executor.memory as bytesConf(ByteUnit.MiB), so a
value without a size
+ // suffix means MiB. Reading it with SparkConf#getSizeAsBytes would treat
8192 as 8192 bytes.
+ val conf = new SparkConf(false).set("spark.executor.memory", "8192")
+ assert(SparkResourceUtil.getExecutorMemorySize(conf) == 8192L * 1024 *
1024)
+ }
+
+ test("getExecutorMemorySize honours a size suffix on spark.executor.memory")
{
+ val conf = new SparkConf(false).set("spark.executor.memory", "8g")
+ assert(SparkResourceUtil.getExecutorMemorySize(conf) == 8L * 1024 * 1024 *
1024)
+ }
+
+ test("getExecutorMemorySize falls back to the Spark default when unset") {
+ // spark.executor.memory defaults to 1g in Spark.
+ val conf = new SparkConf(false)
+ assert(SparkResourceUtil.getExecutorMemorySize(conf) == 1024L * 1024 *
1024)
+ }
+
+ test("getExecutorMemorySize rejects an executor memory that overflows on
conversion") {
+ // The MiB-to-byte conversion multiplies by 2^20, so a suffix-less byte
count large enough to
+ // overflow must fail rather than wrap to a negative budget.
+ val conf = new SparkConf(false).set("spark.executor.memory",
"9000000000000")
+ val e =
intercept[IllegalArgumentException](SparkResourceUtil.getExecutorMemorySize(conf))
+ assert(e.getMessage.contains("exceeds Long.MAX_VALUE"))
+ }
+
+ test("getExecutorMemorySize rejects a negative executor memory") {
+ // Spark's typed entry carries no positivity check, and the conversion
would pass a negative
+ // through, so guard it here rather than propagate a negative budget.
+ val conf = new SparkConf(false).set("spark.executor.memory", "-8192")
+ val e =
intercept[IllegalArgumentException](SparkResourceUtil.getExecutorMemorySize(conf))
+ assert(e.getMessage.contains("spark.executor.memory should not be
negative"))
+ }
+
+ test("getMemoryOverheadSize honours a size suffix on the minimum overhead") {
+ // spark.executor.minMemoryOverhead is a size string, so reading it with
conf.getLong throws
+ // NumberFormatException on any value carrying a unit.
VeloxListenerApi#onDriverStart calls this
+ // unconditionally, so that would abort driver startup on a value Spark
itself accepts.
+ val conf = new SparkConf(false)
+ .set("spark.executor.memory", "1g")
+ .set("spark.executor.minMemoryOverhead", "512m")
+ assert(SparkResourceUtil.getMemoryOverheadSize(conf) == 512L * 1024 * 1024)
+ }
+
+ test("getMemoryOverheadSize falls back to the 384m minimum overhead") {
+ val conf = new SparkConf(false).set("spark.executor.memory", "1g")
+ assert(SparkResourceUtil.getMemoryOverheadSize(conf) == 384L * 1024 * 1024)
+ }
+
+ test("getMemoryOverheadSize prefers the factor when it exceeds the minimum")
{
+ // 8g * 0.1 = 819 MiB, above the 384 MiB floor.
+ val conf = new SparkConf(false)
+ .set("spark.executor.memory", "8g")
+ .set("spark.executor.minMemoryOverhead", "512m")
+ assert(SparkResourceUtil.getMemoryOverheadSize(conf) == 819L * 1024 * 1024)
+ }
+
+ test("getMemoryOverheadSize reads a suffix-less minimum overhead as MiB") {
+ // The value users are most likely to already have. It read the same
before the switch to
+ // getSizeAsMb, so nothing pinned it; a future accessor change could
silently reinterpret it.
+ val conf = new SparkConf(false)
+ .set("spark.executor.memory", "1g")
+ .set("spark.executor.minMemoryOverhead", "512")
+ assert(SparkResourceUtil.getMemoryOverheadSize(conf) == 512L * 1024 * 1024)
+ }
+
+ test("getMemoryOverheadSize rejects an overhead that overflows on
conversion") {
+ // spark.executor.memoryOverhead is MiB-declared and carries no magnitude
check, so the
+ // MiB-to-byte conversion must raise rather than wrap to a negative budget.
+ val conf = new SparkConf(false)
+ .set("spark.executor.memory", "1g")
+ .set("spark.executor.memoryOverhead", "9000000000000")
+ val e =
intercept[IllegalArgumentException](SparkResourceUtil.getMemoryOverheadSize(conf))
+ assert(e.getMessage.contains("exceeds Long.MAX_VALUE"))
+ }
+
+ test("getMemoryOverheadSize rejects a negative explicit overhead") {
+ // An explicit spark.executor.memoryOverhead skips the floor below, so
guard the conversion.
+ val conf = new SparkConf(false)
+ .set("spark.executor.memory", "1g")
+ .set("spark.executor.memoryOverhead", "-1")
+ val e =
intercept[IllegalArgumentException](SparkResourceUtil.getMemoryOverheadSize(conf))
+ assert(e.getMessage.contains("should not be negative"))
+ }
}
diff --git
a/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/GlutenAutoAdjustStageResourceProfile.scala
b/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/GlutenAutoAdjustStageResourceProfile.scala
index 90ecb4cc59..5f80d99dee 100644
---
a/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/GlutenAutoAdjustStageResourceProfile.scala
+++
b/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/GlutenAutoAdjustStageResourceProfile.scala
@@ -32,7 +32,7 @@ import org.apache.spark.sql.execution.adaptive.QueryStageExec
import org.apache.spark.sql.execution.command.{DataWritingCommandExec,
ExecutedCommandExec}
import org.apache.spark.sql.execution.exchange.Exchange
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.util.SparkTestUtil
+import org.apache.spark.util.{SparkResourceUtil, SparkTestUtil}
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
@@ -66,7 +66,8 @@ case class GlutenAutoAdjustStageResourceProfile(glutenConf:
GlutenConfig, spark:
// profile is applied, the settings will be updated accordingly.
GlutenResourceProfile.updateResourceSetting(
ResourceProfile.getOrCreateDefaultProfile(sparkConf),
- sparkConf)
+ sparkConf,
+ isDefaultProfile = true)
if (!plan.isInstanceOf[Exchange]) {
// todo: support set resource profile for final stage
return plan
@@ -183,19 +184,44 @@ object GlutenAutoAdjustStageResourceProfile extends
Logging {
}
/**
- * Reflects resource changes in some configurations that will be passed to
the native side. It
- * only affects the current thread.
+ * Reflects resource changes in some configurations that will be passed to
the native side.
+ *
+ * The values are written into the active SQLConf. On the driver, outside a
task and outside
+ * SQLConf#withExistingConf, that is the session's own conf, so the writes
are visible to every
+ * thread using the session and outlive the query that triggered them.
*/
- def updateResourceSetting(rp: ResourceProfile, sparkConf: SparkConf): Unit =
{
- val coresPerExecutor =
rp.getExecutorCores.getOrElse(sparkConf.get(EXECUTOR_CORES))
- val coresPerTask = rp.getTaskCpus.getOrElse(sparkConf.get(CPUS_PER_TASK))
- val taskSlots = coresPerExecutor / coresPerTask
+ def updateResourceSetting(
+ rp: ResourceProfile,
+ sparkConf: SparkConf,
+ isDefaultProfile: Boolean = false): Unit = {
+ // Resource profiles never take effect in local mode, where a profile
reports
+ // spark.executor.cores (1 by default) rather than the local[N] thread
count that
+ // SparkResourceUtil and GlutenPlugin resolve. Defer to the shared
resolver there so the rule
+ // and the plugin agree on the slot count; elsewhere the profile's own
values are authoritative.
+ val taskSlots = if (SparkResourceUtil.isLocalMaster(sparkConf)) {
+ SparkResourceUtil.getTaskSlots(sparkConf)
+ } else {
+ val coresPerExecutor =
rp.getExecutorCores.getOrElse(sparkConf.get(EXECUTOR_CORES))
+ val coresPerTask = rp.getTaskCpus.getOrElse(sparkConf.get(CPUS_PER_TASK))
+ require(coresPerTask > 0, s"${CPUS_PER_TASK.key} should be positive, but
was $coresPerTask")
+ // Floor at one slot so the division below cannot throw on a combination
Spark itself rejects
+ // later with a dedicated message.
+ Math.max(coresPerExecutor / coresPerTask, 1)
+ }
val conf = SQLConf.get
conf.setConfString(GlutenCoreConfig.NUM_TASK_SLOTS_PER_EXECUTOR.key,
taskSlots.toString)
- val offHeapSize = rp.executorResources
- .get(ResourceProfile.OFFHEAP_MEM)
- .map(_.amount)
- .getOrElse(sparkConf.get(MEMORY_OFFHEAP_SIZE))
+ // A resource profile records executor memory amounts in MiB, while the
two configs written
+ // below are declared as bytesConf(ByteUnit.BYTE). The unmodified default
profile carries the
+ // same off-heap size the conf does, only truncated to MiB, so read the
conf directly there to
+ // keep this in step with what GlutenPlugin wrote at driver init.
+ val offHeapSize = if (isDefaultProfile) {
+ sparkConf.get(MEMORY_OFFHEAP_SIZE)
+ } else {
+ rp.executorResources
+ .get(ResourceProfile.OFFHEAP_MEM)
+ .map(request => SparkResourceUtil.mibToBytes(request.amount))
+ .getOrElse(sparkConf.get(MEMORY_OFFHEAP_SIZE))
+ }
conf.setConfString(GlutenCoreConfig.COLUMNAR_OFFHEAP_SIZE_IN_BYTES.key,
offHeapSize.toString)
conf.setConfString(
GlutenCoreConfig.COLUMNAR_TASK_OFFHEAP_SIZE_IN_BYTES.key,
diff --git
a/gluten-substrait/src/test/scala/org/apache/spark/sql/execution/GlutenAutoAdjustStageResourceProfileSuite.scala
b/gluten-substrait/src/test/scala/org/apache/spark/sql/execution/GlutenAutoAdjustStageResourceProfileSuite.scala
new file mode 100644
index 0000000000..7dc4321261
--- /dev/null
+++
b/gluten-substrait/src/test/scala/org/apache/spark/sql/execution/GlutenAutoAdjustStageResourceProfileSuite.scala
@@ -0,0 +1,129 @@
+/*
+ * 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.sql.execution
+
+import org.apache.gluten.config.GlutenCoreConfig
+
+import org.apache.spark.SparkConf
+import org.apache.spark.resource.{ExecutorResourceRequests, ResourceProfile,
TaskResourceRequests}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.util.SparkResourceUtil
+
+import org.scalatest.funsuite.AnyFunSuite
+
+class GlutenAutoAdjustStageResourceProfileSuite extends AnyFunSuite {
+
+ private val MIB = 1024L * 1024L
+
+ private def clusterConf(offHeap: String): SparkConf = new SparkConf(false)
+ .set("spark.master", "yarn")
+ .set("spark.executor.cores", "4")
+ .set("spark.task.cpus", "1")
+ .set("spark.memory.offHeap.enabled", "true")
+ .set("spark.memory.offHeap.size", offHeap)
+
+ private def profileWith(cores: Int, taskCpus: Int, offHeap: Option[String]):
ResourceProfile = {
+ val ereqs = new ExecutorResourceRequests()
+ ereqs.cores(cores)
+ offHeap.foreach(ereqs.offHeapMemory)
+ val treqs = new TaskResourceRequests()
+ treqs.cpus(taskCpus)
+ new ResourceProfile(ereqs.requests, treqs.requests)
+ }
+
+ test("updateResourceSetting converts the profile's MiB amount to bytes") {
+ // A ResourceProfile records executor memory in MiB, while the configs
written here are declared
+ // as bytesConf(ByteUnit.BYTE). Writing the amount verbatim shrank the
off-heap budget by 2^20,
+ // so 20g became 20480 bytes.
+ val rp = profileWith(cores = 4, taskCpus = 1, offHeap = Some("20g"))
+ SQLConf.withExistingConf(new SQLConf) {
+ GlutenAutoAdjustStageResourceProfile.updateResourceSetting(rp,
clusterConf("20g"))
+ val conf = SQLConf.get
+
assert(conf.getConfString(GlutenCoreConfig.NUM_TASK_SLOTS_PER_EXECUTOR.key) ==
"4")
+ assert(
+
conf.getConfString(GlutenCoreConfig.COLUMNAR_OFFHEAP_SIZE_IN_BYTES.key) ==
+ (20480L * MIB).toString)
+ assert(
+
conf.getConfString(GlutenCoreConfig.COLUMNAR_TASK_OFFHEAP_SIZE_IN_BYTES.key) ==
+ (20480L * MIB / 4).toString)
+ }
+ }
+
+ test("updateResourceSetting reads the exact off-heap size for the default
profile") {
+ // The default profile's amount is the conf value truncated to MiB, so
going through the profile
+ // would shrink a size that is not a whole number of MiB.
spark.memory.offHeap.size=1536k must
+ // stay at 1572864, the value GlutenPlugin already wrote at driver init.
+ val sparkConf = clusterConf("1536k")
+ val rp = ResourceProfile.getOrCreateDefaultProfile(sparkConf)
+ SQLConf.withExistingConf(new SQLConf) {
+ GlutenAutoAdjustStageResourceProfile
+ .updateResourceSetting(rp, sparkConf, isDefaultProfile = true)
+ assert(
+
SQLConf.get.getConfString(GlutenCoreConfig.COLUMNAR_OFFHEAP_SIZE_IN_BYTES.key)
== "1572864")
+ }
+ }
+
+ test("updateResourceSetting agrees with getTaskSlots in local mode") {
+ // A profile reports spark.executor.cores, 1 by default, while
GlutenPlugin resolves local[4] to
+ // four slots. Deriving the count from the profile there made every task
believe it owned the
+ // whole off-heap budget.
+ val sparkConf = new SparkConf(false)
+ .set("spark.master", "local[4]")
+ .set("spark.memory.offHeap.enabled", "true")
+ .set("spark.memory.offHeap.size", "20g")
+ val rp = profileWith(cores = 1, taskCpus = 1, offHeap = Some("20g"))
+ SQLConf.withExistingConf(new SQLConf) {
+ GlutenAutoAdjustStageResourceProfile.updateResourceSetting(rp, sparkConf)
+ val conf = SQLConf.get
+ val expectedSlots = SparkResourceUtil.getTaskSlots(sparkConf)
+ assert(expectedSlots == 4)
+
assert(conf.getConfString(GlutenCoreConfig.NUM_TASK_SLOTS_PER_EXECUTOR.key) ==
"4")
+ assert(
+
conf.getConfString(GlutenCoreConfig.COLUMNAR_TASK_OFFHEAP_SIZE_IN_BYTES.key) ==
+ (20480L * MIB / 4).toString)
+ }
+ }
+
+ test("updateResourceSetting floors the slot count at one") {
+ // spark.task.cpus greater than the executor cores is a combination Spark
rejects later with a
+ // dedicated message, but the division below would throw first on a zero
quotient.
+ val rp = profileWith(cores = 1, taskCpus = 2, offHeap = Some("20g"))
+ SQLConf.withExistingConf(new SQLConf) {
+ GlutenAutoAdjustStageResourceProfile.updateResourceSetting(rp,
clusterConf("20g"))
+ val conf = SQLConf.get
+
assert(conf.getConfString(GlutenCoreConfig.NUM_TASK_SLOTS_PER_EXECUTOR.key) ==
"1")
+ assert(
+
conf.getConfString(GlutenCoreConfig.COLUMNAR_TASK_OFFHEAP_SIZE_IN_BYTES.key) ==
+ (20480L * MIB).toString)
+ }
+ }
+
+ test("updateResourceSetting rejects a non-positive task cpus") {
+ // The profile's own task cpus wins when present, as the two tests above
rely on, so the profile
+ // here carries none and the conf fallback decides.
+ val ereqs = new ExecutorResourceRequests()
+ ereqs.cores(4)
+ ereqs.offHeapMemory("20g")
+ val rp = new ResourceProfile(ereqs.requests, Map.empty)
+ val sparkConf = clusterConf("20g").set("spark.task.cpus", "0")
+ SQLConf.withExistingConf(new SQLConf) {
+ val e = intercept[IllegalArgumentException](
+ GlutenAutoAdjustStageResourceProfile.updateResourceSetting(rp,
sparkConf))
+ assert(e.getMessage.contains("spark.task.cpus should be positive"))
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
