[SPARK-35439][SQL] Children subexpr should come first than parent subexpr

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

@@ -42,7 +42,7 @@ class EquivalentExpressions {
   }
 
   // For each expression, the set of equivalent expressions.
-  private val equivalenceMap = mutable.HashMap.empty[Expr, mutable.ArrayBuffer[Expression]]
+  private val equivalenceMap = mutable.LinkedHashMap.empty[Expr, mutable.ArrayBuffer[Expression]]
 
   /**
    * Adds each expression to this data structure, grouping them with existing equivalent
@@ -167,7 +167,25 @@ class EquivalentExpressions {
    * Returns all the equivalent sets of expressions.
    */
   def getAllEquivalentExprs: Seq[Seq[Expression]] = {
-    equivalenceMap.values.map(_.toSeq).toSeq
+    equivalenceMap.values.map(_.toSeq).toSeq.sortBy(_.head)(new ExpressionOrdering)
+  }
+
+  /**
+   * Orders [Expression] by parent/child relations. The child expression is smaller
+   * than parent expression. If there is child-parent relationships among the subexpressions,
+   * we want the child expressions come first than parent expressions, so we can replace
+   * child expressions in parent expressions with subexpression evaluation.
+   */
+  class ExpressionOrdering extends Ordering[Expression] {
+    override def compare(x: Expression, y: Expression): Int = {
+      if (x.semanticEquals(y)) {
+        0
+      } else if (x.find(_.semanticEquals(y)).isDefined) {
+        1
+      } else {
+        -1
+      }
+    }
   }
 
   /**

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

@@ -309,6 +309,35 @@ class SubexpressionEliminationSuite extends SparkFunSuite with ExpressionEvalHel
       CodeGenerator.compile(code)
     }
   }
+
+  test("SPARK-35439: Children subexpr should come first than parent subexpr") {
+    val add = Add(Literal(1), Literal(2))
+
+    val equivalence1 = new EquivalentExpressions
+
+    equivalence1.addExprTree(add)
+    assert(equivalence1.getAllEquivalentExprs.head === Seq(add))
+
+    equivalence1.addExprTree(Add(Literal(3), add))
+    assert(equivalence1.getAllEquivalentExprs ===
+      Seq(Seq(add, add), Seq(Add(Literal(3), add))))
+
+    equivalence1.addExprTree(Add(Literal(3), add))
+    assert(equivalence1.getAllEquivalentExprs ===
+      Seq(Seq(add, add), Seq(Add(Literal(3), add), Add(Literal(3), add))))
+
+    val equivalence2 = new EquivalentExpressions
+
+    equivalence2.addExprTree(Add(Literal(3), add))
+    assert(equivalence2.getAllEquivalentExprs === Seq(Seq(add), Seq(Add(Literal(3), add))))
+
+    equivalence2.addExprTree(add)
+    assert(equivalence2.getAllEquivalentExprs === Seq(Seq(add, add), Seq(Add(Literal(3), add))))
+
+    equivalence2.addExprTree(Add(Literal(3), add))
+    assert(equivalence2.getAllEquivalentExprs ===
+      Seq(Seq(add, add), Seq(Add(Literal(3), add), Add(Literal(3), add))))
+  }
 }
 
 case class CodegenFallbackExpression(child: Expression)