dwsmith1983 commented on code in PR #6547:
URL: https://github.com/apache/datafusion-comet/pull/6547#discussion_r4171276599
##########
spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala:
##########
@@ -608,6 +610,89 @@ case class CometExecRule(session: SparkSession)
}
}
+ /**
+ * AQE re-plans around a materialized stage by reusing the physical node
linked to it, so a
+ * native operator that shares its logical node with a shuffle stage (the
final aggregate of a
+ * two-phase aggregate) keeps the native plan it got while that input was a
bare exchange, read
+ * through a plain `Scan`. Once the input is a sink that reads the shuffle
directly, its
+ * `ShuffleScan` takes the place of the stale leaf. The leaf is patched in
place because
+ * converting the node again from `originalPlan` would drop the stage's
logical link that AQE
+ * relies on and re-run serde on a node that is already planned.
+ */
+ private def refreshStaleShuffleScans(op: SparkPlan): SparkPlan = op match {
+ // These build their own `Scan` over the child rather than embedding the
child's plan.
+ case _: CometNativeWriteExec | _: CometIcebergWriteExec | _:
CometWriteFilesExec => op
Review Comment:
Added `scansChildAsSeparateBlock`, used here and in the `firstNativeOp`
reset in `_apply`.
--
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]
