Rewrite the code block that compares the equivalency of

Seq[Expression] in semanticEquals.

Author: codingjaguar

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

@@ -48,7 +48,7 @@ import org.apache.spark.sql.types._
  *                       the same output data type.
  *
  */
-abstract class Expression extends TreeNode[Expression] with PredicateHelper{
+abstract class Expression extends TreeNode[Expression]{
 
   /**
    * Returns true when an expression is a candidate for static evaluation before the query is
@@ -160,6 +160,18 @@ abstract class Expression extends TreeNode[Expression] with PredicateHelper{
     checkSemantic(elements1, elements2)
   }
 
+  /**
+   * Returns a sequence of expressions by removing from q the first expression that is semantically
+   * equivalent to e.
+   */
+  def removeFirstSemanticEquivalent(seq: Seq[Expression], e: Expression): Seq[Expression] = {
+    seq match {
+      case Seq() => Seq()
+      case x +: rest if x semanticEquals e => rest
+      case x +: rest => x +: removeFirstSemanticEquivalent(rest, e)
+    }
+  }
+
   /**
    * Returns the hash for this expression. Expressions that compute the same result, even if
    * they differ cosmetically should return the same hash.

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

@@ -228,7 +228,8 @@ case class InSet(child: Expression, hset: Set[Any]) extends UnaryExpression with
   }
 }
 
-case class And(left: Expression, right: Expression) extends BinaryOperator with Predicate {
+case class And(left: Expression, right: Expression) extends BinaryOperator
+  with Predicate with PredicateHelper{
 
   override def inputType: AbstractDataType = BooleanType
 
@@ -256,10 +257,12 @@ case class And(left: Expression, right: Expression) extends BinaryOperator with
     // Non-deterministic expressions cannot be semantic equal
     if (!deterministic || !other.deterministic) return false
 
-    // we know both expressions are And, so we can tolerate ordering different
-    val elements1 = splitConjunctivePredicates(this).toSet.toSeq
-    val elements2 = splitConjunctivePredicates(other).toSet.toSeq
-    checkSemantic(elements1, elements2)
+    // We already know both expressions are And, so we can tolerate ordering different
+    // Recursively call semanticEquals on subexpressions to check the equivalency of two seqs.
+    var elements1 = splitConjunctivePredicates(this)
+    val elements2 = splitConjunctivePredicates(other)
+    for (e <- elements2) elements1 = removeFirstSemanticEquivalent(elements1, e)
+    elements1.isEmpty
   }
 
   override def genCode(ctx: CodeGenContext, ev: GeneratedExpressionCode): String = {
@@ -287,7 +290,8 @@ case class And(left: Expression, right: Expression) extends BinaryOperator with
 }
 
 
-case class Or(left: Expression, right: Expression) extends BinaryOperator with Predicate {
+case class Or(left: Expression, right: Expression) extends BinaryOperator
+  with Predicate with PredicateHelper {
 
   override def inputType: AbstractDataType = BooleanType
 
@@ -316,9 +320,11 @@ case class Or(left: Expression, right: Expression) extends BinaryOperator with P
     if (!deterministic || !other.deterministic) return false
 
     // we know both expressions are Or, so we can tolerate ordering different
-    val elements1 = splitDisjunctivePredicates(this).toSet.toSeq
-    val elements2 = splitDisjunctivePredicates(other).toSet.toSeq
-    checkSemantic(elements1, elements2)
+    // Recursively call semanticEquals on subexpressions to check the equivalency of two seqs.
+    var elements1 = splitDisjunctivePredicates(this)
+    val elements2 = splitDisjunctivePredicates(other)
+    for (e <- elements2) elements1 = removeFirstSemanticEquivalent(elements1, e)
+    elements1.isEmpty
   }
 
   override def genCode(ctx: CodeGenContext, ev: GeneratedExpressionCode): String = {

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlan.scala

@@ -25,7 +25,7 @@ import org.apache.spark.sql.catalyst.plans.QueryPlan
 import org.apache.spark.sql.catalyst.trees.{CurrentOrigin, TreeNode}
 
 
-abstract class LogicalPlan extends QueryPlan[LogicalPlan] with PredicateHelper with Logging {
+abstract class LogicalPlan extends QueryPlan[LogicalPlan] with Logging {
 
   private var _analyzed: Boolean = false
 

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/SameResultSuite.scala

@@ -63,6 +63,9 @@ class SameResultSuite extends SparkFunSuite {
     assertSameResult(testRelation.where('a === 'b || 'c === 'd),
       testRelation2.where('c === 'd || 'a === 'b )
     )
+    assertSameResult(testRelation.where(('a === 'b || 'c === 'd) && ('e === 'f || 'g === 'h)),
+      testRelation2.where(('g === 'h || 'e === 'f) && ('c === 'd || 'a === 'b ))
+    )
 
     assertSameResult(testRelation.where('a === 'b && 'c === 'd),
       testRelation2.where('a === 'c && 'b === 'd),