(kyuubi) branch master updated: [KYUUBI #6211] Check memory offHeap enabled for CustomResourceProfileExec

[wangzhen]

This is an automated email from the ASF dual-hosted git repository.

wangzhen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kyuubi.git


The following commit(s) were added to refs/heads/master by this push:
     new 9114e507c [KYUUBI #6211] Check memory offHeap enabled for 
CustomResourceProfileExec
9114e507c is described below

commit 9114e507c41f7a8efb96032e98d5d41f41f67f8e
Author: wforget <643348...@qq.com>
AuthorDate: Thu Mar 28 13:17:59 2024 +0800

    [KYUUBI #6211] Check memory offHeap enabled for CustomResourceProfileExec
    
    # :mag: Description
    ## Issue References 🔗
    
    This pull request fixes #
    
    ## Describe Your Solution 🔧
    
    We should check `spark.memory.offHeap.enabled` when applying for 
`executorOffHeapMemory`.
    
    ## Types of changes :bookmark:
    
    - [X] Bugfix (non-breaking change which fixes an issue)
    - [ ] New feature (non-breaking change which adds functionality)
    - [ ] Breaking change (fix or feature that would cause existing 
functionality to change)
    
    ## Test Plan 🧪
    
    #### Behavior Without This Pull Request :coffin:
    
    #### Behavior With This Pull Request :tada:
    
    #### Related Unit Tests
    
    ---
    
    # Checklist 📝
    
    - [X] This patch was not authored or co-authored using [Generative 
Tooling](https://www.apache.org/legal/generative-tooling.html)
    
    **Be nice. Be informative.**
    
    Closes #6211 from wForget/hotfix.
    
    Closes #6211
    
    1c7c8cd75 [wforget] Check memory offHeap enabled for 
CustomResourceProfileExec
    
    Authored-by: wforget <643348...@qq.com>
    Signed-off-by: wforget <643348...@qq.com>
---
 .../org/apache/spark/sql/execution/CustomResourceProfileExec.scala | 7 ++++++-
 .../org/apache/spark/sql/execution/CustomResourceProfileExec.scala | 7 ++++++-
 .../org/apache/spark/sql/execution/CustomResourceProfileExec.scala | 7 ++++++-
 3 files changed, 18 insertions(+), 3 deletions(-)

diff --git 
a/extensions/spark/kyuubi-extension-spark-3-3/src/main/scala/org/apache/spark/sql/execution/CustomResourceProfileExec.scala
 
b/extensions/spark/kyuubi-extension-spark-3-3/src/main/scala/org/apache/spark/sql/execution/CustomResourceProfileExec.scala
index 3698140fb..043d6496b 100644
--- 
a/extensions/spark/kyuubi-extension-spark-3-3/src/main/scala/org/apache/spark/sql/execution/CustomResourceProfileExec.scala
+++ 
b/extensions/spark/kyuubi-extension-spark-3-3/src/main/scala/org/apache/spark/sql/execution/CustomResourceProfileExec.scala
@@ -53,7 +53,12 @@ case class CustomResourceProfileExec(child: SparkPlan) 
extends UnaryExecNode {
   private val executorMemoryOverhead =
     conf.getConf(FINAL_WRITE_STAGE_EXECUTOR_MEMORY_OVERHEAD)
       .getOrElse(sparkContext.getConf.get("spark.executor.memoryOverhead", 
"1G"))
-  private val executorOffHeapMemory = 
conf.getConf(FINAL_WRITE_STAGE_EXECUTOR_OFF_HEAP_MEMORY)
+  private val executorOffHeapMemory =
+    if (sparkContext.getConf.getBoolean("spark.memory.offHeap.enabled", 
false)) {
+      conf.getConf(FINAL_WRITE_STAGE_EXECUTOR_OFF_HEAP_MEMORY)
+    } else {
+      None
+    }
 
   override lazy val metrics: Map[String, SQLMetric] = {
     val base = Map(
diff --git 
a/extensions/spark/kyuubi-extension-spark-3-4/src/main/scala/org/apache/spark/sql/execution/CustomResourceProfileExec.scala
 
b/extensions/spark/kyuubi-extension-spark-3-4/src/main/scala/org/apache/spark/sql/execution/CustomResourceProfileExec.scala
index 3698140fb..043d6496b 100644
--- 
a/extensions/spark/kyuubi-extension-spark-3-4/src/main/scala/org/apache/spark/sql/execution/CustomResourceProfileExec.scala
+++ 
b/extensions/spark/kyuubi-extension-spark-3-4/src/main/scala/org/apache/spark/sql/execution/CustomResourceProfileExec.scala
@@ -53,7 +53,12 @@ case class CustomResourceProfileExec(child: SparkPlan) 
extends UnaryExecNode {
   private val executorMemoryOverhead =
     conf.getConf(FINAL_WRITE_STAGE_EXECUTOR_MEMORY_OVERHEAD)
       .getOrElse(sparkContext.getConf.get("spark.executor.memoryOverhead", 
"1G"))
-  private val executorOffHeapMemory = 
conf.getConf(FINAL_WRITE_STAGE_EXECUTOR_OFF_HEAP_MEMORY)
+  private val executorOffHeapMemory =
+    if (sparkContext.getConf.getBoolean("spark.memory.offHeap.enabled", 
false)) {
+      conf.getConf(FINAL_WRITE_STAGE_EXECUTOR_OFF_HEAP_MEMORY)
+    } else {
+      None
+    }
 
   override lazy val metrics: Map[String, SQLMetric] = {
     val base = Map(
diff --git 
a/extensions/spark/kyuubi-extension-spark-3-5/src/main/scala/org/apache/spark/sql/execution/CustomResourceProfileExec.scala
 
b/extensions/spark/kyuubi-extension-spark-3-5/src/main/scala/org/apache/spark/sql/execution/CustomResourceProfileExec.scala
index 3698140fb..043d6496b 100644
--- 
a/extensions/spark/kyuubi-extension-spark-3-5/src/main/scala/org/apache/spark/sql/execution/CustomResourceProfileExec.scala
+++ 
b/extensions/spark/kyuubi-extension-spark-3-5/src/main/scala/org/apache/spark/sql/execution/CustomResourceProfileExec.scala
@@ -53,7 +53,12 @@ case class CustomResourceProfileExec(child: SparkPlan) 
extends UnaryExecNode {
   private val executorMemoryOverhead =
     conf.getConf(FINAL_WRITE_STAGE_EXECUTOR_MEMORY_OVERHEAD)
       .getOrElse(sparkContext.getConf.get("spark.executor.memoryOverhead", 
"1G"))
-  private val executorOffHeapMemory = 
conf.getConf(FINAL_WRITE_STAGE_EXECUTOR_OFF_HEAP_MEMORY)
+  private val executorOffHeapMemory =
+    if (sparkContext.getConf.getBoolean("spark.memory.offHeap.enabled", 
false)) {
+      conf.getConf(FINAL_WRITE_STAGE_EXECUTOR_OFF_HEAP_MEMORY)
+    } else {
+      None
+    }
 
   override lazy val metrics: Map[String, SQLMetric] = {
     val base = Map(