diff --git a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala index 9888eaa179..2aaf5586c3 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala @@ -144,10 +144,10 @@ object VeloxRuleApi { injector.injectFinal(c => RemoveGlutenTableCacheColumnarToRow(c.session)) injector.injectFinal( c => PreventBatchTypeMismatchInTableCache(c.caller.isCache(), Set(VeloxBatchType))) - injector.injectFinal( - c => GlutenAutoAdjustStageResourceProfile(new GlutenConfig(c.sqlConf), c.session)) injector.injectFinal( c => AdjustStageExecutionMode(new GlutenConfig(c.sqlConf), c.session, c.caller.isAqe())) + injector.injectFinal( + c => GlutenAutoAdjustStageResourceProfile(new GlutenConfig(c.sqlConf), c.session)) injector.injectFinal(c => GlutenFallbackReporter(new GlutenConfig(c.sqlConf), c.session)) injector.injectFinal(_ => RemoveFallbackTagRule()) } diff --git a/backends-velox/src/main/scala/org/apache/spark/sql/execution/AdjustStageExecutionMode.scala b/backends-velox/src/main/scala/org/apache/spark/sql/execution/AdjustStageExecutionMode.scala index adadc67df7..1cd89ac610 100644 --- a/backends-velox/src/main/scala/org/apache/spark/sql/execution/AdjustStageExecutionMode.scala +++ b/backends-velox/src/main/scala/org/apache/spark/sql/execution/AdjustStageExecutionMode.scala @@ -27,6 +27,7 @@ import org.apache.spark.sql.catalyst.rules.Rule import org.apache.spark.sql.catalyst.trees.TreeNodeTag import org.apache.spark.sql.execution.AdjustStageExecutionMode.{adjustExecutionMode, unsetTag} import org.apache.spark.sql.execution.adaptive.{AQEShuffleReadExec, ColumnarAQEShuffleReadExec, ShuffleQueryStageExec} +import org.apache.spark.sql.execution.joins.BaseJoinExec import org.apache.spark.util.SparkTestUtil // For ShuffleStage, the resource profile is set to ColumnarShuffleExchangeExec.inputColumnarRDD. @@ -51,30 +52,19 @@ case class AdjustStageExecutionMode( } private def adjustExecutionModeForGPU(plan: SparkPlan): SparkPlan = { - val transformers = plan.collect { case t: WholeStageTransformer => t } - if (transformers.isEmpty) { - return plan - } - if (transformers.size > 1) { - logWarning(s"Not offloading GPU because multiple WholeStageTransformer exist. Remove tags.") + if (!AdjustStageExecutionMode.offloadGpu(plan, glutenConf)) { unsetTag(plan, CudfTag.CudfTag) return plan } - val transformer = transformers.head - if (transformer.isCudf) { - val gpuStageMode = if (SparkTestUtil.isTesting) { - // Only unset for transformer. - transformer.unsetTagValue(CudfTag.CudfTag) - MockGPUStageMode - } else { - GPUStageMode - } - - adjustExecutionMode(plan, gpuStageMode) + val gpuStageMode = if (SparkTestUtil.isTesting) { + unsetTag(plan, CudfTag.CudfTag) + MockGPUStageMode } else { - plan + GPUStageMode } + + adjustExecutionMode(plan, gpuStageMode) } } @@ -97,9 +87,9 @@ object AdjustStageExecutionMode extends Logging { shuffle .copy(mapperStageMode = Some(stageExecutionMode)) .withNewChildren(Seq(adjustExecutionMode(shuffle.child, stageExecutionMode))) - case resizeBatches: VeloxResizeBatchesExec => + case VeloxResizeBatchesExec(child @ AQEShuffleReadExec(_, _), _) => VeloxResizeBatchesExec( - adjustExecutionMode(resizeBatches.child, stageExecutionMode), + adjustExecutionMode(child, stageExecutionMode), Some(stageExecutionMode)) case _ => plan.withNewChildren(plan.children.map(adjustExecutionMode(_, stageExecutionMode))) @@ -113,4 +103,33 @@ object AdjustStageExecutionMode extends Logging { case _ => } } + + def offloadGpu(plan: SparkPlan, glutenConf: GlutenConfig): Boolean = { + if (glutenConf.gpuOnlyOffloadJoinStage) { + if (!plan.exists(_.isInstanceOf[BaseJoinExec])) { + logWarning(s"Not offloading GPU because missing offload condition.") + return false + } + } + + val transformers = plan.collect { case t: WholeStageTransformer => t } + + if (transformers.isEmpty) { + logWarning(s"Not offloading GPU because no WholeStageTransformer.") + return false + } + + if (transformers.size > 1) { + // Do not offload GPU if the whole stage is broken down into multiple native pipelines. + logWarning(s"Not offloading GPU because multiple WholeStageTransformer exist.") + return false + } + + if (!transformers.head.isCudf) { + logWarning(s"Not offloading GPU because WholeStageTransformer is not tagged cudf.") + return false + } + + true + } } diff --git a/backends-velox/src/test/scala/org/apache/gluten/execution/StageExecutionModeSuite.scala b/backends-velox/src/test/scala/org/apache/gluten/execution/StageExecutionModeSuite.scala index 50458d43b6..ee21c1ad93 100644 --- a/backends-velox/src/test/scala/org/apache/gluten/execution/StageExecutionModeSuite.scala +++ b/backends-velox/src/test/scala/org/apache/gluten/execution/StageExecutionModeSuite.scala @@ -113,20 +113,17 @@ class StageExecutionModeSuite extends VeloxWholeStageTransformerSuite { s"Expected GPU AQE shuffle reader, but got ${reader.executionMode}") } - val shuffleStages = plan.collect { - case stage: ShuffleQueryStageExec => stage + val shuffleStages = shuffleReaders.map(_.delegate).map { + case Left(aqeReader) => + assert(aqeReader.child.isInstanceOf[ShuffleQueryStageExec]) + aqeReader.child.asInstanceOf[ShuffleQueryStageExec] + case Right(shuffleQueryStageExec: ShuffleQueryStageExec) => shuffleQueryStageExec } - val exchanges = shuffleStages.flatMap { - _.plan.collect { - case exchange: ColumnarShuffleExchangeExec => exchange - } - } - - assert(exchanges.nonEmpty) - - exchanges.foreach { - exchange => + shuffleStages.foreach { + shuffleStage => + assert(shuffleStage.shuffle.isInstanceOf[ColumnarShuffleExchangeExec]) + val exchange = shuffleStage.shuffle.asInstanceOf[ColumnarShuffleExchangeExec] assert( !exchange.mapperStageMode.contains(MockGPUStageMode), s"Expected CPU mapper stage, but got ${exchange.mapperStageMode}") diff --git a/docs/Configuration.md b/docs/Configuration.md index 83f44ec941..36a6745a38 100644 --- a/docs/Configuration.md +++ b/docs/Configuration.md @@ -152,13 +152,18 @@ nav_order: 15 ## Gluten *experimental* configurations -| Key | Modifiability | Default | Description | -|-------------------------------------------------------------------|---------------|---------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| -| spark.gluten.auto.adjustStageResource.enabled | 🔄 Dynamic | false | Experimental: If enabled, gluten will try to set the stage resource according to stage execution plan. Only worked when aqe is enabled at the same time!! | -| spark.gluten.auto.adjustStageResources.fallenNode.ratio.threshold | 🔄 Dynamic | 0.5 | Experimental: Increase executor heap memory when stage contains fallen node count exceeds the total node count ratio. | -| spark.gluten.auto.adjustStageResources.heap.ratio | 🔄 Dynamic | 2.0 | Experimental: Increase executor heap memory when match adjust stage resource rule. | -| spark.gluten.auto.adjustStageResources.offheap.ratio | 🔄 Dynamic | 0.5 | Experimental: Decrease executor offheap memory when match adjust stage resource rule. | -| spark.gluten.memory.dynamic.offHeap.sizing.enabled | ⚓ Static | false | Experimental: When set to true, the offheap config (spark.memory.offHeap.size) will be ignored and instead we will consider onheap and offheap memory in combination, both counting towards the executor memory config (spark.executor.memory). We will make use of JVM APIs to determine how much onheap memory is use, alongside tracking offheap allocations made by Gluten. We will then proceed to enforcing a total memory quota, calculated by the sum of what memory is committed and in use in the Java heap. Since the calculation of the total quota happens as offheap allocation happens and not as JVM heap memory is allocated, it is possible that we can oversubscribe memory. Additionally, note that this change is experimental and may have performance implications. | -| spark.gluten.memory.dynamic.offHeap.sizing.memory.fraction | ⚓ Static | 0.6 | Experimental: Determines the memory fraction used to determine the total memory available for offheap and onheap allocations when the dynamic offheap sizing feature is enabled. The default is set to match spark.executor.memoryFraction. | -| spark.gluten.sql.columnar.cudf | 🔄 Dynamic | false | Enable or disable cudf support. This is an experimental feature. | +| Key | Modifiability | Default | Description | +|---------------------------------------------------------------------|---------------|---------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| spark.gluten.auto.adjustStageResource.enabled | 🔄 Dynamic | false | Experimental: If enabled, gluten will try to set the stage resource according to stage execution plan. Only worked when aqe is enabled at the same time!! | +| spark.gluten.auto.adjustStageResources.fallenNode.ratio.threshold | 🔄 Dynamic | 0.5 | Experimental: Increase executor heap memory when stage contains fallen node count exceeds the total node count ratio. | +| spark.gluten.auto.adjustStageResources.heap.ratio | 🔄 Dynamic | 2.0 | Experimental: Increase executor heap memory when match adjust stage resource rule. | +| spark.gluten.auto.adjustStageResources.offheap.ratio | 🔄 Dynamic | 0.5 | Experimental: Decrease executor offheap memory when match adjust stage resource rule. | +| spark.gluten.memory.dynamic.offHeap.sizing.enabled | ⚓ Static | false | Experimental: When set to true, the offheap config (spark.memory.offHeap.size) will be ignored and instead we will consider onheap and offheap memory in combination, both counting towards the executor memory config (spark.executor.memory). We will make use of JVM APIs to determine how much onheap memory is use, alongside tracking offheap allocations made by Gluten. We will then proceed to enforcing a total memory quota, calculated by the sum of what memory is committed and in use in the Java heap. Since the calculation of the total quota happens as offheap allocation happens and not as JVM heap memory is allocated, it is possible that we can oversubscribe memory. Additionally, note that this change is experimental and may have performance implications. | +| spark.gluten.memory.dynamic.offHeap.sizing.memory.fraction | ⚓ Static | 0.6 | Experimental: Determines the memory fraction used to determine the total memory available for offheap and onheap allocations when the dynamic offheap sizing feature is enabled. The default is set to match spark.executor.memoryFraction. | +| spark.gluten.sql.columnar.cudf | 🔄 Dynamic | false | Enable or disable cudf support. This is an experimental feature. | +| spark.gluten.sql.columnar.gpu.onlyOffloadJoinStage | 🔄 Dynamic | false | If true, Gluten will only offload join stages to GPU. Other stages will be executed on CPU. | +| spark.gluten.sql.columnar.hybridExecution.cpuResource.name | ⚓ Static | cpu | The CPU resource name. This is used to schedule non-GPU tasks to target nodes with the resource name. | +| spark.gluten.sql.columnar.hybridExecution.enabled | ⚓ Static | false | Enable CPU/GPU hybrid execution. At runtime, the execution will be scheduled to target nodes based on the selected execution mode. | +| spark.gluten.sql.columnar.hybridExecution.gpuResource.amountPerTask | ⚓ Static | 1.0 | The GPU resource amount per task. This is used to limit GPU tasks to target nodes. | +| spark.gluten.sql.columnar.hybridExecution.gpuResource.name | ⚓ Static | gpu | The GPU resource name. This is used to schedule GPU tasks to target nodes with the resource name. | diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala b/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala index 98fe6a0657..8f76c11f75 100644 --- a/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala +++ b/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala @@ -396,6 +396,14 @@ class GlutenConfig(conf: SQLConf) extends GlutenCoreConfig(conf) { def maxBroadcastTableSize: Long = JavaUtils.byteStringAsBytes(conf.getConfString(SPARK_MAX_BROADCAST_TABLE_SIZE, "8GB")) + + def enableHybridExecution: Boolean = getConf(ENABLE_HYBRID_EXECUTION) + + def cpuResourceName: String = getConf(HYBRID_EXECUTION_CPU_RESOURCE_NAME) + def gpuResourceName: String = getConf(HYBRID_EXECUTION_GPU_RESOURCE_NAME) + def gpuResourceAmountPerTask: Double = getConf(HYBRID_EXECUTION_GPU_RESOURCE_AMOUNT_PER_TASK) + + def gpuOnlyOffloadJoinStage: Boolean = getConf(GPU_ONLY_OFFLOAD_JOIN_STAGE) } object GlutenConfig extends ConfigRegistry { @@ -1713,4 +1721,48 @@ object GlutenConfig extends ConfigRegistry { "total size of small files is below this threshold.") .doubleConf .createWithDefault(0.5) + + val ENABLE_HYBRID_EXECUTION = + buildStaticConf("spark.gluten.sql.columnar.hybridExecution.enabled") + .experimental() + .doc( + "Enable CPU/GPU hybrid execution. At runtime, the execution will be scheduled to target " + + "nodes based on the selected execution mode.") + .booleanConf + .createWithDefault(false) + + val HYBRID_EXECUTION_CPU_RESOURCE_NAME = + buildStaticConf("spark.gluten.sql.columnar.hybridExecution.cpuResource.name") + .experimental() + .doc( + "The CPU resource name. This is used to schedule non-GPU tasks to target nodes with " + + "the resource name.") + .stringConf + .createWithDefault("cpu") + + val HYBRID_EXECUTION_GPU_RESOURCE_NAME = + buildStaticConf("spark.gluten.sql.columnar.hybridExecution.gpuResource.name") + .experimental() + .doc( + "The GPU resource name. This is used to schedule GPU tasks to target nodes with " + + "the resource name.") + .stringConf + .createWithDefault("gpu") + + val HYBRID_EXECUTION_GPU_RESOURCE_AMOUNT_PER_TASK = + buildStaticConf("spark.gluten.sql.columnar.hybridExecution.gpuResource.amountPerTask") + .experimental() + .doc( + "The GPU resource amount per task. This is used to limit GPU tasks to target nodes.") + .doubleConf + .createWithDefault(1.0) + + val GPU_ONLY_OFFLOAD_JOIN_STAGE = + buildConf("spark.gluten.sql.columnar.gpu.onlyOffloadJoinStage") + .experimental() + .doc( + "If true, Gluten will only offload join stages to GPU." + + " Other stages will be executed on CPU.") + .booleanConf + .createWithDefault(false) } 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..e1be285020 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 @@ -17,7 +17,7 @@ package org.apache.spark.sql.execution import org.apache.gluten.config.{GlutenConfig, GlutenCoreConfig} -import org.apache.gluten.execution.{ColumnarToRowExecBase, GlutenPlan} +import org.apache.gluten.execution.{ColumnarToRowExecBase, GlutenPlan, WholeStageTransformer} import org.apache.gluten.logging.LogLevelUtil import org.apache.spark.SparkConf @@ -67,6 +67,28 @@ case class GlutenAutoAdjustStageResourceProfile(glutenConf: GlutenConfig, spark: GlutenResourceProfile.updateResourceSetting( ResourceProfile.getOrCreateDefaultProfile(sparkConf), sparkConf) + + val rpManager = spark.sparkContext.resourceProfileManager + val defaultRP = rpManager.defaultResourceProfile + + // initial resource profile config as default resource profile + val taskResource = mutable.Map.empty[String, TaskResourceRequest] ++= defaultRP.taskResources + val executorResource = + mutable.Map.empty[String, ExecutorResourceRequest] ++= defaultRP.executorResources + + if (glutenConf.enableColumnarCudf && glutenConf.enableHybridExecution) { + val transformers = plan.collect { case t: WholeStageTransformer => t } + if (transformers.nonEmpty && transformers.forall(_.isCudf)) { + return GlutenResourceProfile.setResourceProfileForGpu( + plan, + executorResource, + taskResource, + rpManager, + sparkConf, + glutenConf) + } + } + if (!plan.isInstanceOf[Exchange]) { // todo: support set resource profile for final stage return plan @@ -77,25 +99,18 @@ case class GlutenAutoAdjustStageResourceProfile(glutenConf: GlutenConfig, spark: } log.info(s"detailPlanNodes ${planNodes.map(_.nodeName).mkString("Array(", ", ", ")")}") - // one stage is considered as fallback if all node is not GlutenPlan - // or all GlutenPlan node is C2R node. - val wholeStageFallback = planNodes - .filter(_.isInstanceOf[GlutenPlan]) - .count(!_.isInstanceOf[ColumnarToRowExecBase]) == 0 - - val rpManager = spark.sparkContext.resourceProfileManager - val defaultRP = rpManager.defaultResourceProfile - - // initial resource profile config as default resource profile - val taskResource = mutable.Map.empty[String, TaskResourceRequest] ++= defaultRP.taskResources - val executorResource = - mutable.Map.empty[String, ExecutorResourceRequest] ++= defaultRP.executorResources val memoryRequest = executorResource.get(ResourceProfile.MEMORY) val offheapRequest = executorResource.get(ResourceProfile.OFFHEAP_MEM) logInfo(s"default memory request $memoryRequest") logInfo(s"default offheap request $offheapRequest") // case 1: whole stage fallback to vanilla spark in such case we increase the heap + // + // one stage is considered as fallback if all node is not GlutenPlan + // or all GlutenPlan node is C2R node. + val wholeStageFallback = planNodes + .filter(_.isInstanceOf[GlutenPlan]) + .count(!_.isInstanceOf[ColumnarToRowExecBase]) == 0 if (wholeStageFallback) { val newMemoryAmount = memoryRequest.get.amount * glutenConf.autoAdjustStageRPHeapRatio val newExecutorMemory = @@ -106,10 +121,10 @@ case class GlutenAutoAdjustStageResourceProfile(glutenConf: GlutenConfig, spark: new ExecutorResourceRequest(ResourceProfile.OFFHEAP_MEM, offheapRequest.get.amount / 10) executorResource.put(ResourceProfile.OFFHEAP_MEM, newExecutorOffheap) - val newRP = new ResourceProfile(executorResource.toMap, taskResource.toMap) - return GlutenResourceProfile.applyNewResourceProfileIfPossible( + return GlutenResourceProfile.applyNewResourceProfile( plan, - newRP, + executorResource, + taskResource, rpManager, sparkConf) } @@ -131,10 +146,10 @@ case class GlutenAutoAdjustStageResourceProfile(glutenConf: GlutenConfig, spark: new ExecutorResourceRequest(ResourceProfile.OFFHEAP_MEM, newOffHeapMemoryAmount.toLong) executorResource.put(ResourceProfile.OFFHEAP_MEM, newExecutorOffheap) - val newRP = new ResourceProfile(executorResource.toMap, taskResource.toMap) - return GlutenResourceProfile.applyNewResourceProfileIfPossible( + return GlutenResourceProfile.applyNewResourceProfile( plan, - newRP, + executorResource, + taskResource, rpManager, sparkConf) } @@ -202,17 +217,47 @@ object GlutenAutoAdjustStageResourceProfile extends Logging { (offHeapSize / taskSlots).toString) } - def applyNewResourceProfileIfPossible( + def applyNewResourceProfile( plan: SparkPlan, - rp: ResourceProfile, + executorResource: mutable.Map[String, ExecutorResourceRequest], + taskResource: mutable.Map[String, TaskResourceRequest], rpManager: ResourceProfileManager, sparkConf: SparkConf): SparkPlan = { - updateResourceSetting(rp, sparkConf) - + val rp = new ResourceProfile(executorResource.toMap, taskResource.toMap) val finalRP = getFinalResourceProfile(rpManager, rp) - // Wrap the plan with ApplyResourceProfileExec so that we can apply new ResourceProfile - val wrapperPlan = ApplyResourceProfileExec(plan.children.head, finalRP) - logInfo(s"Apply resource profile $finalRP for plan ${wrapperPlan.nodeName}") - plan.withNewChildren(IndexedSeq(wrapperPlan)) + updateResourceSetting(finalRP, sparkConf) + + plan match { + case shuffle: Exchange => + logInfo(s"Apply resource profile $finalRP for plan ${shuffle.child.nodeName}") + // Wrap the plan with ApplyResourceProfileExec so that we can apply new ResourceProfile + val wrapperPlan = ApplyResourceProfileExec(shuffle.child, finalRP) + shuffle.withNewChildren(Seq(wrapperPlan)) + case other => + logInfo(s"Apply resource profile $finalRP for plan ${other.nodeName}") + ApplyResourceProfileExec(other, finalRP) + } + } + + def setResourceProfileForGpu( + plan: SparkPlan, + executorResource: mutable.Map[String, ExecutorResourceRequest], + taskResource: mutable.Map[String, TaskResourceRequest], + rpManager: ResourceProfileManager, + sparkConf: SparkConf, + glutenConf: GlutenConfig): SparkPlan = { + // The gpu task resource limits how many tasks can be launched in one executor. + val gpuResourceName = glutenConf.gpuResourceName + taskResource.put( + gpuResourceName, + new TaskResourceRequest(gpuResourceName, glutenConf.gpuResourceAmountPerTask)) + executorResource.put(gpuResourceName, new ExecutorResourceRequest(gpuResourceName, 1)) + executorResource.remove(glutenConf.cpuResourceName) + applyNewResourceProfile( + plan, + executorResource, + taskResource, + rpManager, + sparkConf) } }