peter-toth commented on a change in pull request #26044: [SPARK-29375][SQL]
Exchange reuse across all subquery levels
URL: https://github.com/apache/spark/pull/26044#discussion_r334267924
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/Exchange.scala
##########
@@ -107,35 +107,22 @@ case class ReuseExchange(conf: SQLConf) extends
Rule[SparkPlan] {
if (!conf.exchangeReuseEnabled) {
return plan
}
- // Build a hash map using schema of exchanges to avoid O(N*N) sameResult
calls.
- val exchanges = mutable.HashMap[StructType, ArrayBuffer[Exchange]]()
+ val exchanges = mutable.HashMap[SparkPlan, Exchange]()
- // Replace a Exchange duplicate with a ReusedExchange
- def reuse: PartialFunction[Exchange, SparkPlan] = {
+ def reuse(plan: SparkPlan): SparkPlan = plan.transform {
case exchange: Exchange =>
- val sameSchema = exchanges.getOrElseUpdate(exchange.schema,
ArrayBuffer[Exchange]())
- val samePlan = sameSchema.find { e =>
- exchange.sameResult(e)
- }
- if (samePlan.isDefined) {
- // Keep the output of this exchange, the following plans require
that to resolve
- // attributes.
- ReusedExchangeExec(exchange.output, samePlan.get)
+ val newExchange = exchanges.getOrElseUpdate(exchange.canonicalized,
exchange)
Review comment:
I see your point. Let me change this PR a bit and come back to you.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
