[SPARK-35742][SQL] Expression.semanticEquals should be symmetrical

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Expression.scala

@@ -238,7 +238,7 @@ abstract class Expression extends TreeNode[Expression] {
    *
    * See [[Canonicalize]] for more details.
    */
-  def semanticEquals(other: Expression): Boolean =
+  final def semanticEquals(other: Expression): Boolean =
     deterministic && other.deterministic && canonicalized == other.canonicalized
 
   /**

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/namedExpressions.scala

@@ -278,11 +278,6 @@ case class AttributeReference(
     case _ => false
   }
 
-  override def semanticEquals(other: Expression): Boolean = other match {
-    case ar: AttributeReference => sameRef(ar)
-    case _ => false
-  }
-
   override def semanticHash(): Int = {
     this.exprId.hashCode()
   }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/subquery.scala

@@ -76,13 +76,6 @@ abstract class SubqueryExpression(
     AttributeSet.fromAttributeSets(outerAttrs.map(_.references))
   override def children: Seq[Expression] = outerAttrs ++ joinCond
   override def withNewPlan(plan: LogicalPlan): SubqueryExpression
-  override def semanticEquals(o: Expression): Boolean = o match {
-    case p: SubqueryExpression =>
-      this.getClass.getName.equals(p.getClass.getName) && plan.sameResult(p.plan) &&
-        children.length == p.children.length &&
-        children.zip(p.children).forall(p => p._1.semanticEquals(p._2))
-    case _ => false
-  }
 }
 
 object SubqueryExpression {

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CanonicalizeSuite.scala

@@ -170,4 +170,11 @@ class CanonicalizeSuite extends SparkFunSuite {
       assert(nestedExpr2.canonicalized != nestedExpr3.canonicalized)
     }
   }
+
+  test("SPARK-35742: Expression.semanticEquals should be symmetrical") {
+    val attr = AttributeReference("col", IntegerType)()
+    val expr = PromotePrecision(attr)
+    assert(expr.semanticEquals(attr))
+    assert(attr.semanticEquals(expr))
+  }
 }

sql/core/src/main/scala/org/apache/spark/sql/execution/SubqueryAdaptiveBroadcastExec.scala

@@ -20,6 +20,7 @@ package org.apache.spark.sql.execution
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.QueryPlan
 
 /**
  * Similar to [[SubqueryBroadcastExec]], this node is used to store the
@@ -40,6 +41,11 @@ case class SubqueryAdaptiveBroadcastExec(
       "SubqueryAdaptiveBroadcastExec does not support the execute() code path.")
   }
 
+  protected override def doCanonicalize(): SparkPlan = {
+    val keys = buildKeys.map(k => QueryPlan.normalizeExpressions(k, child.output))
+    copy(name = "dpp", buildKeys = keys, child = child.canonicalized)
+  }
+
   override protected def withNewChildInternal(newChild: SparkPlan): SubqueryAdaptiveBroadcastExec =
     copy(child = newChild)
 }

sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala

@@ -73,7 +73,9 @@ case class HashAggregateExec(
   // This is for testing. We force TungstenAggregationIterator to fall back to the unsafe row hash
   // map and/or the sort-based aggregation once it has processed a given number of input rows.
   private val testFallbackStartsAt: Option[(Int, Int)] = {
-    sqlContext.getConf("spark.sql.TungstenAggregate.testFallbackStartsAt", null) match {
+    Option(sqlContext).map { sc =>
+      sc.getConf("spark.sql.TungstenAggregate.testFallbackStartsAt", null)
+    }.orNull match {
       case null | "" => None
       case fallbackStartsAt =>
         val splits = fallbackStartsAt.split(",").map(_.trim)

sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala

@@ -71,9 +71,8 @@ case class ScalarSubquery(
   override def toString: String = plan.simpleString(SQLConf.get.maxToStringFields)
   override def withNewPlan(query: BaseSubqueryExec): ScalarSubquery = copy(plan = query)
 
-  override def semanticEquals(other: Expression): Boolean = other match {
-    case s: ScalarSubquery => plan.sameResult(s.plan)
-    case _ => false
+  override lazy val canonicalized: Expression = {
+    ScalarSubquery(plan.canonicalized.asInstanceOf[BaseSubqueryExec], ExprId(0))
   }
 
   // the first column in first row from `query`.
@@ -127,11 +126,6 @@ case class InSubqueryExec(
   override def withNewPlan(plan: BaseSubqueryExec): InSubqueryExec = copy(plan = plan)
   final override def nodePatternsInternal: Seq[TreePattern] = Seq(IN_SUBQUERY_EXEC)
 
-  override def semanticEquals(other: Expression): Boolean = other match {
-    case in: InSubqueryExec => child.semanticEquals(in.child) && plan.sameResult(in.plan)
-    case _ => false
-  }
-
   def updateResult(): Unit = {
     val rows = plan.executeCollect()
     result = if (plan.output.length > 1) {