[GLUTEN-12676][CORE] Read MiB-declared Spark memory configs with matching accessors - #12677
Open
LuciferYang wants to merge 5 commits into
Open
[GLUTEN-12676][CORE] Read MiB-declared Spark memory configs with matching accessors#12677LuciferYang wants to merge 5 commits into
LuciferYang wants to merge 5 commits into
Conversation
The dynamic off-heap sizing branch read spark.executor.memory with SparkConf#getSizeAsBytes, which treats a value without a size suffix as bytes. Spark declares the config as bytesConf(ByteUnit.MiB), so a bare value means MiB. With spark.executor.memory=8192 the branch saw 8192 bytes instead of 8 GiB, and (onHeapSize - 300MiB) * 0.6 turned the off-heap budget negative. Add SparkResourceUtil.getExecutorMemorySize, which reads the typed EXECUTOR_MEMORY entry and converts from MiB, matching how getMemoryOverheadSize in the same object already reads it. The typed entry also carries Spark's 1g default, so the hand-rolled conf.contains check and hardcoded 1GB fallback are no longer needed.
Reading spark.executor.memory as MiB fixed one negative-budget path but opened another: ByteUnit#toBytes only rejects negatives, then multiplies by 2^20 with no overflow check, so a suffix-less byte count at or above 2^43 wraps. With spark.executor.memory=9000000000000 and dynamic sizing the driver started and spark.gluten.memory.offHeap.size.in.bytes came out as -5405736044414474240, where the parent commit produced a positive value. Convert with ByteUnit#convertTo, which raises on overflow, and require a non-negative value first, since the typed EXECUTOR_MEMORY entry carries no positivity check and convertTo passes negatives through. Add GlutenDynamicOffHeapSizingSuite, which boots a SparkContext through spark.plugins and asserts the budget derived from a suffix-less spark.executor.memory. The existing tests only cover the helper, so reverting the call site alone kept them green. Also drop GlutenCoreConfig.SPARK_ONHEAP_SIZE_KEY, whose last reader this change removed, and narrow the helper's scaladoc to what holds for the helper itself: the MiB semantics, plus the note that standalone and local-cluster resolve the config through SparkContext#executorMemoryInMb and treat a suffix-less value as bytes.
Two more sites carry the same unit mismatch the executor-memory read had. getMemoryOverheadSize reads spark.executor.minMemoryOverhead with conf.getLong, but that config is a size string, MiB unless suffixed, so String.toLong fails on any value carrying a unit. VeloxListenerApi#onDriverStart calls the method unconditionally, and isMemoryOverheadSet counts minMemoryOverhead, so setting it guarantees the raw read is reached: spark.executor.minMemoryOverhead=512m aborts driver startup with NumberFormatException on a value Spark itself accepts, and docs/velox-spark-configuration.md lists that config for users. getSizeAsMb keeps the result in MiB, which is the unit the surrounding max compares against, and works on every supported Spark version, unlike the 4.0-only EXECUTOR_MIN_MEMORY_OVERHEAD entry. GlutenAutoAdjustStageResourceProfile#updateResourceSetting takes the off-heap amount from a ResourceProfile, which records executor memory in MiB, and wrote it verbatim into two configs declared as bytesConf(ByteUnit.BYTE). With spark.memory.offHeap.size=20g the session's off-heap budget became 20480 bytes instead of 21474836480, and the per-task budget 5120 bytes. The MEMORY_OFFHEAP_SIZE fallback on the same expression is already in bytes, so the two branches disagreed by 2^20. Convert with ByteUnit#convertTo, which raises on overflow rather than wrapping. Both are reachable through configuration alone: the first needs only a suffixed minMemoryOverhead, the second spark.gluten.auto.adjustStageResource.enabled with AQE and spark.memory.offHeap.enabled, all three of which the rule checks before it does anything.
Follow-ups from reviewing the previous commit. SparkResourceUtil gains a mibToBytes helper holding the sign check and the overflow-raising convertTo, and the three MiB-to-byte boundaries all go through it. getMemoryOverheadSize previously ended in ByteUnit.MiB.toBytes, which wraps silently past 2^43 MiB, so spark.executor.minMemoryOverhead=9000000t produced a negative overhead budget. A bare toBytes-to-convertTo swap would have regressed the other end: convertTo passes negatives through, and an explicit spark.executor.memoryOverhead=-1 skips the .max floor below, so the require stays inside the helper. updateResourceSetting held a second implementation of getTaskSlots that disagreed with it in local mode: getTaskSlots resolves local[8] to 8 while a profile reports spark.executor.cores, 1 by default. Before the unit fix that divergence made every per-task budget 2^20 times too small, which masked it; with the total corrected it becomes an 8x over-provision, each of 8 concurrent tasks believing it owns the whole off-heap budget. Local mode now defers to the shared resolver, and the non-local branch gains the positivity check and the one-slot floor that getTaskSlots already carries, so a profile with CORES=1 and spark.task.cpus=2 no longer throws "/ by zero" on every query. The off-heap read also splits: the unmodified default profile carries the conf value truncated to MiB, so spark.memory.offHeap.size=1536k came back as 1048576 instead of the 1572864 the plugin wrote at driver init. The default-profile call site now says so with an isDefaultProfile flag and reads the conf directly. Deriving the flag from rp.id does not work: nextProfileId starts at 0, so the first profile constructed in a JVM gets id 0 whether or not it is the default. Tests: three for the overhead conversion (suffix-less MiB, overflow, negative) and three for the rule (local-mode slot agreement, the slot floor, the non-positive task cpus rejection). The test that exercised the OFFHEAP_MEM getOrElse fallback is replaced by one covering the default-profile branch, since neither call site can reach that fallback in production.
|
Run Gluten Clickhouse CI on x86 |
Contributor
There was a problem hiding this comment.
Pull request overview
Fixes incorrect interpretation of Spark memory configs whose declared units (MiB vs bytes) didn’t match how Gluten read/wrote them, leading to wrong (sometimes negative) off-heap budgets. The changes centralize MiB→bytes conversion with overflow/sign guarding, align ResourceProfile-based values with byte-based Gluten configs, and add regression coverage for the affected paths.
Changes:
- Read
spark.executor.memoryandspark.executor.minMemoryOverheadusing accessors consistent with their MiB-based semantics and guard MiB→bytes conversions against negatives/overflow. - Fix
ResourceProfileoff-heap amounts (MiB) being written into byte-based Gluten configs; align task-slot calculation withSparkResourceUtil.getTaskSlots, especially forlocal[N]. - Add targeted unit/integration-style tests covering unit handling, overflow/negative rejection, default-profile exact-byte behavior, and local-mode slot agreement.
Reviewed changes
Copilot reviewed 7 out of 7 changed files in this pull request and generated no comments.
Show a summary per file
| File | Description |
|---|---|
| gluten-core/src/main/scala/org/apache/spark/util/SparkResourceUtil.scala | Adds getExecutorMemorySize and a shared mibToBytes helper; fixes min overhead parsing via getSizeAsMb. |
| gluten-core/src/main/scala/org/apache/gluten/GlutenPlugin.scala | Switches dynamic sizing to use SparkResourceUtil.getExecutorMemorySize instead of getSizeAsBytes/manual default. |
| gluten-core/src/main/scala/org/apache/gluten/config/GlutenCoreConfig.scala | Removes the unused SPARK_ONHEAP_SIZE_KEY constant. |
| gluten-substrait/src/main/scala/org/apache/spark/sql/execution/GlutenAutoAdjustStageResourceProfile.scala | Fixes MiB-vs-bytes mismatch when writing off-heap sizes; aligns task slots with shared resolver in local mode; adds validation/flooring. |
| gluten-core/src/test/scala/org/apache/spark/util/SparkResourceUtilSuite.scala | Adds tests for executor memory/min overhead parsing, defaults, overflow, and negative rejection. |
| gluten-core/src/test/scala/org/apache/gluten/GlutenDynamicOffHeapSizingSuite.scala | Adds a regression test that drives SparkContext + plugin init for suffix-less executor memory dynamic sizing. |
| gluten-substrait/src/test/scala/org/apache/spark/sql/execution/GlutenAutoAdjustStageResourceProfileSuite.scala | Adds tests for MiB→bytes conversion, default-profile exact-byte read, local-mode slot agreement, and task-cpus validation. |
Suppressed comments (1)
gluten-core/src/main/scala/org/apache/gluten/GlutenPlugin.scala:120
- In the dynamic off-heap sizing branch, the first warning is triggered by a user-defined
spark.memory.offHeap.enabled(SPARK_OFFHEAP_ENABLED_KEY) but the message incorrectly claims it is ignoring the off-heap size key. This makes troubleshooting confusing and duplicates the next warning, which already covers the size key.
if (conf.contains(GlutenCoreConfig.SPARK_OFFHEAP_ENABLED_KEY)) {
logWarning(
s"Dynamic off-heap sizing is enabled. Ignoring user-defined " +
s"'${GlutenCoreConfig.SPARK_OFFHEAP_SIZE_KEY}' setting.")
}
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
|
Run Gluten Clickhouse CI on x86 |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
Three Spark memory configurations are declared with a unit but read through an accessor that assumes a different one, so the off-heap budgets Gluten derives from them are wrong. Each commit fixes one site, and the last one cleans up two problems the fixes exposed.
spark.executor.memoryis declaredbytesConf(ByteUnit.MiB), so a value without a suffix means MiB. The dynamic off-heap sizing branch read it withSparkConf#getSizeAsBytes, which reads a suffix-less value as bytes. Withspark.executor.memory=8192it saw 8192 bytes instead of 8 GiB, and(onHeapSize - 300MiB) * 0.6made the off-heap budget negative.SparkResourceUtil.getExecutorMemorySizenow reads the typedEXECUTOR_MEMORYentry and converts from MiB, which is also howgetMemoryOverheadSizein the same object already read it. The typed entry carries Spark's 1g default, so the hand-rolledconf.containscheck and the hardcoded 1GB fallback are gone.spark.executor.minMemoryOverheadis a size string, MiB unless suffixed.getMemoryOverheadSizeread it withconf.getLong, which isString.toLongand fails on anything carrying a unit.VeloxListenerApi#onDriverStartcalls the method unconditionally andisMemoryOverheadSetcountsminMemoryOverhead, so setting it guarantees the raw read is reached:spark.executor.minMemoryOverhead=512maborted driver startup withNumberFormatExceptionon a value Spark itself accepts, anddocs/velox-spark-configuration.mdlists that config for users.getSizeAsMbkeeps the result in MiB, which is the unit the surroundingmaxcompares against, and it works on every supported Spark version, unlike the 4.0-onlyEXECUTOR_MIN_MEMORY_OVERHEADentry.A
ResourceProfilerecords executor memory amounts in MiB, whilespark.gluten.memory.offHeap.size.in.bytesandspark.gluten.memory.task.offHeap.size.in.bytesare declaredbytesConf(ByteUnit.BYTE).GlutenAutoAdjustStageResourceProfile#updateResourceSettingwrote the profile amount verbatim, so withspark.memory.offHeap.size=20gthe session off-heap budget became 20480 bytes instead of 21474836480, and the per-task budget 5120 bytes. TheMEMORY_OFFHEAP_SIZEfallback on the same expression is already in bytes, so the two branches of one expression disagreed by 2^20.ByteUnit#toBytesrejects a negative input but multiplies by 2^20 with no overflow check, so a suffix-less byte count at or above 2^43 wraps.spark.executor.memory=9000000000000with dynamic sizing started the driver and producedspark.gluten.memory.offHeap.size.in.bytes=-5405736044414474240.SparkResourceUtilnow has onemibToByteshelper holding arequire(>= 0)and the overflow-raisingconvertTo, and all three MiB-to-byte boundaries go through it. A baretoBytestoconvertToswap would have regressed the other end, sinceconvertTopasses negatives through and an explicitspark.executor.memoryOverhead=-1skips the.maxfloor, so therequirelives inside the helper.updateResourceSettingalso held a second implementation ofSparkResourceUtil.getTaskSlotsthat disagreed with it in local mode:getTaskSlotsresolveslocal[8]to 8 slots, while a profile reportsspark.executor.cores, 1 by default. While the total was also 2^20 times too small the two errors cancelled, which is why nobody noticed. Correct the total and it becomes an 8x over-provision, with each of 8 concurrent tasks sized as if it owned the whole budget. Local mode now defers to the shared resolver, and the other branch gains the positivity check and one-slot floor thatgetTaskSlotsalready carries, so a profile withCORES=1andspark.task.cpus=2no longer throws/ by zeroon every query. The off-heap read splits the same way: the unmodified default profile carries the conf value truncated to MiB, sospark.memory.offHeap.size=1536kcame back as 1048576 instead of the 1572864 the plugin wrote at driver init, and the default-profile call site now says so with anisDefaultProfileflag and reads the conf directly. Deriving that flag fromrp.iddoes not work:nextProfileIdstarts at 0, so the first profile constructed in a JVM gets id 0 whether or not it is the default.The second commit also drops
GlutenCoreConfig.SPARK_ONHEAP_SIZE_KEY, whose last reader the first commit removed.How was this patch tested?
SparkResourceUtilSuitecovers the three accessors:spark.executor.memoryas a bare value and with a suffix, the 1g default, overflow and negative rejection,minMemoryOverheadwith a suffix and without, the 384m floor, the factor path, and overflow and negative rejection onmemoryOverhead.GlutenAutoAdjustStageResourceProfileSuitecovers the rule: the MiB-to-byte conversion, the exact-bytes read for the default profile, agreement withgetTaskSlotsunderlocal[4], the one-slot floor, and the non-positivespark.task.cpusrejection.GlutenDynamicOffHeapSizingSuiteboots aSparkContextthroughspark.pluginsand asserts the budget derived from a suffix-lessspark.executor.memory, because the helper-level tests stayed green when only the call site was reverted.Every test was checked against the unfixed code first: reverting each fix individually turns the matching tests red.
mvn -Pspark-3.5 -pl gluten-core,gluten-substrait testgives 51 and 56 passing.-Pspark-3.3compiles both modules; on-Pspark-4.1 -Pscala-2.13gluten-corecompiles andgluten-substraithas pre-existing failures unrelated to this change.Was this patch authored or co-authored using generative AI tooling?
No
Closes #12676