ulysses-you commented on code in PR #57602:
URL: https://github.com/apache/spark/pull/57602#discussion_r3689634309
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala:
##########
@@ -802,10 +802,25 @@ abstract class SparkStrategies extends
QueryPlanner[SparkPlan] {
object WindowGroupLimit extends Strategy {
def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
case logical.WindowGroupLimit(partitionSpec, orderSpec,
rankLikeFunction, limit, child) =>
- val partialWindowGroupLimit =
execution.window.WindowGroupLimitExec(partitionSpec,
- orderSpec, rankLikeFunction, limit, execution.window.Partial,
planLater(child))
+ // When the partial window group limit is bypassed, skip the
pre-shuffle partial
+ // WindowGroupLimit and run only a single WindowGroupLimit after the
shuffle. This can
+ // improve performance when the pre-shuffle reduction ratio is low.
+ //
+ // The bypass is only gated on a non-empty partitionSpec. With an
empty partitionSpec the
+ // final WindowGroupLimit requires AllTuples, so the shuffle funnels
the whole input into a
+ // single reducer; the partial pass we would drop bounds each input
partition to `limit`
+ // rank groups before that shuffle, so it always helps and should
never be skipped.
Review Comment:
addressed
--
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]
