dongjoon-hyun commented on code in PR #56928:
URL: https://github.com/apache/spark/pull/56928#discussion_r3510596539
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/ExpandExec.scala:
##########
@@ -43,8 +43,17 @@ case class ExpandExec(
"numOutputRows" -> SQLMetrics.createMetric(sparkContext, "number of output
rows"))
// The GroupExpressions can output data with arbitrary partitioning, so set
it
- // as UNKNOWN partitioning
- override def outputPartitioning: Partitioning = UnknownPartitioning(0)
+ // as UNKNOWN partitioning. Expand only replicates rows within a partition
and never moves rows
+ // across partitions, so when the single-task optimization is enabled and
the child produces a
+ // single partition, we can forward the `SinglePartition` property to avoid
an unneeded shuffle.
+ override def outputPartitioning: Partitioning = {
+ if (conf.getConf(SQLConf.SINGLE_TASK_EXECUTION_EXPAND) &&
Review Comment:
- If we read this session configuration every time, AQE seems to make this
logic unstably.
- Is this applied for all `ExpandExec` (both marked and unmarked)?
--
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]
