[SPARK-35855][SQL] Unify reuse map data structures in non-AQE and AQE rules

### What changes were proposed in this pull request?
This PR unifies reuse map data structures in non-AQE and AQE rules to a simple `Map[<canonicalized plan>, <plan>]` based on the discussion here: #28885 (comment)

### Why are the changes needed?
The proposed `Map[<canonicalized plan>, <plan>]` is simpler than the currently used `Map[<schema>, ArrayBuffer[<plan>]]` in `ReuseMap`/`ReuseExchangeAndSubquery` (non-AQE) and consistent with the `ReuseAdaptiveSubquery` (AQE) subquery reuse rule.

### Does this PR introduce _any_ user-facing change?
No.

### How was this patch tested?
Existing UTs.

Closes #33021 from peter-toth/SPARK-35855-unify-reuse-map-data-structures.

Authored-by: Peter Toth <peter.toth@gmail.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>

Author: peter-toth

sql/catalyst/src/main/scala/org/apache/spark/sql/util/ReuseMap.scala

@@ -17,11 +17,12 @@
 
 package org.apache.spark.sql.execution.reuse
 
+import scala.collection.mutable
+
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.trees.TreePattern._
 import org.apache.spark.sql.execution.{BaseSubqueryExec, ExecSubqueryExpression, ReusedSubqueryExec, SparkPlan}
 import org.apache.spark.sql.execution.exchange.{Exchange, ReusedExchangeExec}
-import org.apache.spark.sql.util.ReuseMap
 
 /**
  * Find out duplicated exchanges and subqueries in the whole spark plan including subqueries, then
@@ -36,24 +37,34 @@ case object ReuseExchangeAndSubquery extends Rule[SparkPlan] {
 
   def apply(plan: SparkPlan): SparkPlan = {
     if (conf.exchangeReuseEnabled || conf.subqueryReuseEnabled) {
-      val exchanges = new ReuseMap[Exchange, SparkPlan]()
-      val subqueries = new ReuseMap[BaseSubqueryExec, SparkPlan]()
+      val exchanges = mutable.Map.empty[SparkPlan, Exchange]
+      val subqueries = mutable.Map.empty[SparkPlan, BaseSubqueryExec]
 
       def reuse(plan: SparkPlan): SparkPlan = {
         plan.transformUpWithPruning(_.containsAnyPattern(EXCHANGE, PLAN_EXPRESSION)) {
           case exchange: Exchange if conf.exchangeReuseEnabled =>
-            exchanges.reuseOrElseAdd(exchange, ReusedExchangeExec(exchange.output, _))
+            val cachedExchange = exchanges.getOrElseUpdate(exchange.canonicalized, exchange)
+            if (cachedExchange.ne(exchange)) {
+              ReusedExchangeExec(exchange.output, cachedExchange)
+            } else {
+              cachedExchange
+            }
 
           case other =>
             other.transformExpressionsUpWithPruning(_.containsPattern(PLAN_EXPRESSION)) {
               case sub: ExecSubqueryExpression =>
                 val subquery = reuse(sub.plan).asInstanceOf[BaseSubqueryExec]
-                sub.withNewPlan(
-                  if (conf.subqueryReuseEnabled) {
-                    subqueries.reuseOrElseAdd(subquery, ReusedSubqueryExec(_))
+                val newSubquery = if (conf.subqueryReuseEnabled) {
+                  val cachedSubquery = subqueries.getOrElseUpdate(subquery.canonicalized, subquery)
+                  if (cachedSubquery.ne(subquery)) {
+                    ReusedSubqueryExec(cachedSubquery)
                   } else {
-                    subquery
-                  })
+                    cachedSubquery
+                  }
+                } else {
+                  subquery
+                }
+                sub.withNewPlan(newSubquery)
             }
         }
       }

sql/catalyst/src/test/scala/org/apache/spark/sql/util/ReuseMapSuite.scala

@@ -479,9 +479,9 @@ abstract class BroadcastJoinSuiteBase extends QueryTest with SQLTestUtils
   test("broadcast join where streamed side's output partitioning is PartitioningCollection") {
     withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500") {
       val t1 = (0 until 100).map(i => (i % 5, i % 13)).toDF("i1", "j1")
-      val t2 = (0 until 100).map(i => (i % 5, i % 13)).toDF("i2", "j2")
+      val t2 = (0 until 100).map(i => (i % 5, i % 14)).toDF("i2", "j2")
       val t3 = (0 until 20).map(i => (i % 7, i % 11)).toDF("i3", "j3")
-      val t4 = (0 until 100).map(i => (i % 5, i % 13)).toDF("i4", "j4")
+      val t4 = (0 until 100).map(i => (i % 5, i % 15)).toDF("i4", "j4")
 
       // join1 is a sort merge join (shuffle on the both sides).
       val join1 = t1.join(t2, t1("i1") === t2("i2"))