jerrypeng commented on code in PR #59523:
URL: https://github.com/apache/spark/pull/59523#discussion_r4429446712
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnablePipelinedShuffle.scala:
##########
@@ -127,24 +131,26 @@ object EnablePipelinedShuffle extends Rule[SparkPlan] {
* True if any [[ShuffleExchangeExec]] in `plan` is read by an operator the
channel transport
* cannot serve. The unsupported operators (see class doc for why each is
fatal) are
* `CoalesceExec`, `CartesianProductExec`, and the limit operators
`CollectLimitExec` /
- * `CollectTailExec` / `TakeOrderedAndProjectExec`. For each such operator
anywhere in the plan,
- * check whether a shuffle is reachable below it.
- *
- * The reachability walk descends through EVERY child of a non-exchange node
-- not only unary
- * children -- so a shuffle behind a `UnionExec`/join (a `BinaryExecNode`)
beneath the operator
- * is still found. It stops at the FIRST [[ShuffleExchangeExec]] on each
path: a shuffle deeper
- * than that first one is not read by this operator (the intervening
exchange's own reader reads
- * one reduce partition per task), so it is not this operator's concern.
+ * `CollectTailExec` / `TakeOrderedAndProjectExec`, plus cached
`AttachDistributedSequenceExec`.
+ * A shuffle above an unsupported operator with shuffle-free input does not
disqualify the plan.
*/
- private def readsShuffleThroughUnsupportedConsumer(plan: SparkPlan): Boolean
= {
- def reachesShuffle(p: SparkPlan): Boolean = p match {
- case _: ShuffleExchangeExec => true
- case other => other.children.exists(reachesShuffle)
- }
- plan.exists {
- case p if PipelinedShuffleEligibility.isUnsupportedConsumer(p) =>
- p.children.exists(reachesShuffle)
- case _ => false
+ private[exchange] def readsShuffleThroughUnsupportedConsumer(plan:
SparkPlan): Boolean = {
Review Comment:
Agreed that the captured mutable flag and manual iterator added unnecessary
complexity. I replaced them with a top-down traversal using `children.exists`
and an immutable flag indicating an unsupported ancestor on the current path.
Reaching a shuffle with that flag set returns true; sibling branches keep
independent state.
This preserves early exit and linear traversal work. I retained the semantic
tests and reduced the counted-work grid to one depth-128 regression. That test
guards against repeated subtree scans; it does not establish a measured
planning-latency improvement.
--
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]
