Copilot commented on code in PR #12588:
URL: https://github.com/apache/gluten/pull/12588#discussion_r3710577974
##########
gluten-substrait/src/main/scala/org/apache/spark/sql/execution/adaptive/ColumnarAQEShuffleReadExec.scala:
##########
@@ -31,40 +31,56 @@ import org.apache.spark.sql.vectorized.ColumnarBatch
* ShuffleQueryStageExec if executionMode is set by the planner.
*
* @param delegate
- * The AQEShuffleReadExec or ShuffleQueryStageExec.
+ * AQEShuffleReadExec or ShuffleQueryStageExec. Or ShuffleExchange during
canonicalization.
* @param executionMode
* The execution mode of the current AQE stage.
*/
case class ColumnarAQEShuffleReadExec(
- delegate: Either[AQEShuffleReadExec, ShuffleQueryStageExec],
+ delegate: SparkPlan,
executionMode: StageExecutionMode) extends UnaryExecNode {
override def nodeName: String =
s"ColumnarAQEShuffleRead(${executionMode.name})"
- private val isAQEShuffleRead = delegate.isLeft
-
- private val aqeReader: AQEShuffleReadExec = {
- if (isAQEShuffleRead) {
- delegate.left.get
- } else {
- // Wrap ShuffleQueryStageExe with dummy PartitionSpecs.
- val queryStageExec = delegate.right.get
- // Create CoalescedPartitionSpec for each partition.
- val partitionSpecs =
- Array.tabulate(queryStageExec.shuffle.numPartitions)(i =>
CoalescedPartitionSpec(i, i + 1))
- AQEShuffleReadExec(queryStageExec, partitionSpecs)
- }
+ override def supportsColumnar: Boolean = true
+
+ override def child: SparkPlan = delegate match {
+ case AQEShuffleReadExec(c, _) => c
+ case _ => delegate
}
- override def supportsColumnar: Boolean = true
+ override def output: Seq[Attribute] = delegate.output
- override def child: SparkPlan = aqeReader.child
+ override lazy val outputPartitioning: Partitioning =
delegate.outputPartitioning
- override def output: Seq[Attribute] = aqeReader.child.output
+ override protected def stringArgs: Iterator[Any] = {
+ delegate match {
+ case a: AQEShuffleReadExec => a.stringArgs
+ case _ => super.stringArgs
+ }
+ }
- override lazy val outputPartitioning: Partitioning =
aqeReader.outputPartitioning
+ override protected def withNewChildInternal(newChild: SparkPlan):
ColumnarAQEShuffleReadExec = {
+ delegate match {
+ case a: AQEShuffleReadExec => copy(delegate =
a.withNewChildren(Seq(newChild)))
+ case _ => copy(delegate = newChild)
+ }
+ }
- override def stringArgs: Iterator[Any] = aqeReader.stringArgs
+ private lazy val aqeReader: AQEShuffleReadExec = {
+ delegate match {
+ case a: AQEShuffleReadExec => a
+ case s: ShuffleQueryStageExec =>
+ // Wrap ShuffleQueryStageExe with dummy PartitionSpecs by creating
CoalescedPartitionSpec
+ // for each partition.
Review Comment:
Typo in comment: `ShuffleQueryStageExe` should be `ShuffleQueryStageExec`.
##########
gluten-substrait/src/main/scala/org/apache/spark/sql/execution/adaptive/ColumnarAQEShuffleReadExec.scala:
##########
@@ -31,40 +31,56 @@ import org.apache.spark.sql.vectorized.ColumnarBatch
* ShuffleQueryStageExec if executionMode is set by the planner.
*
* @param delegate
- * The AQEShuffleReadExec or ShuffleQueryStageExec.
+ * AQEShuffleReadExec or ShuffleQueryStageExec. Or ShuffleExchange during
canonicalization.
* @param executionMode
* The execution mode of the current AQE stage.
*/
case class ColumnarAQEShuffleReadExec(
- delegate: Either[AQEShuffleReadExec, ShuffleQueryStageExec],
+ delegate: SparkPlan,
executionMode: StageExecutionMode) extends UnaryExecNode {
override def nodeName: String =
s"ColumnarAQEShuffleRead(${executionMode.name})"
- private val isAQEShuffleRead = delegate.isLeft
-
- private val aqeReader: AQEShuffleReadExec = {
- if (isAQEShuffleRead) {
- delegate.left.get
- } else {
- // Wrap ShuffleQueryStageExe with dummy PartitionSpecs.
- val queryStageExec = delegate.right.get
- // Create CoalescedPartitionSpec for each partition.
- val partitionSpecs =
- Array.tabulate(queryStageExec.shuffle.numPartitions)(i =>
CoalescedPartitionSpec(i, i + 1))
- AQEShuffleReadExec(queryStageExec, partitionSpecs)
- }
+ override def supportsColumnar: Boolean = true
+
+ override def child: SparkPlan = delegate match {
+ case AQEShuffleReadExec(c, _) => c
+ case _ => delegate
}
- override def supportsColumnar: Boolean = true
+ override def output: Seq[Attribute] = delegate.output
- override def child: SparkPlan = aqeReader.child
+ override lazy val outputPartitioning: Partitioning =
delegate.outputPartitioning
Review Comment:
outputPartitioning is taken from `delegate`, which is incorrect when
`delegate` is a `ShuffleQueryStageExec`: `aqeReader` wraps it with synthetic
partitionSpecs, and those specs affect the partitioning semantics. Returning
`delegate.outputPartitioning` can therefore diverge from the actual shuffle
reader behavior.
This issue also appears on line 62 of the same file.
##########
gluten-substrait/src/main/scala/org/apache/spark/sql/execution/GlutenAutoAdjustStageResourceProfile.scala:
##########
@@ -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)
+ }
+ }
Review Comment:
GPU resource profile assignment runs before the `Exchange` guard, which
contradicts this rule’s own comment/TODO about not supporting the final stage
yet. If this rule is applied to a non-Exchange root plan, it can wrap the whole
plan in `ApplyResourceProfileExec`, changing semantics from “per-stage” to
“whole query”.
##########
gluten-substrait/src/main/scala/org/apache/spark/sql/execution/GlutenAutoAdjustStageResourceProfile.scala:
##########
@@ -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)
Review Comment:
setResourceProfileForGpu removes the CPU resource only from
`executorResource` but leaves any CPU requirement in `taskResource`. If the
default profile has a CPU task resource requirement, GPU stages will still
require that CPU resource and can become unschedulable on GPU-only nodes.
##########
gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala:
##########
@@ -1715,4 +1723,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")
Review Comment:
The doc for `spark.gluten.sql.columnar.hybridExecution.cpuResource.name`
suggests this setting alone schedules non-GPU tasks, but Gluten only uses it to
remove the CPU resource from GPU stages. Consider clarifying that it must match
a Spark custom resource configured via `spark.executor.resource.<name>.*` /
`spark.task.resource.<name>.*` for scheduling to take effect.
##########
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.
|
Review Comment:
This table entry implies the Gluten setting alone schedules tasks to CPU
nodes, but the implementation relies on Spark custom resource configuration
(executor/task resources) and only removes the CPU resource for GPU stages.
Please clarify the description to avoid suggesting this works without Spark
resource configs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
