[SPARK-12161][SQL] Ignore order of predicates in cache matching

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

@@ -65,6 +65,20 @@ trait PredicateHelper {
     }
   }
 
+  /**
+   * Returns true if two expressions are equivalent predicates. Equality check is tolerant of
+   * ordering different.
+   */
+  def equivalentPredicates(left: Expression, right: Expression): Boolean = {
+    val leftAndPredicates = splitConjunctivePredicates(left).toSet
+    val rightAndPredicates = splitConjunctivePredicates(right).toSet
+    val leftOrPredicates = splitDisjunctivePredicates(left).toSet
+    val rightOrPredicates = splitDisjunctivePredicates(right).toSet
+    // We split the two conditions into conjunctive predicates and disjunctive predicates
+    // If either of them match, we consider them equivalent conditions
+    (leftAndPredicates == rightAndPredicates) || (leftOrPredicates == rightOrPredicates)
+  }
+
   // Substitute any known alias from a map.
   protected def replaceAlias(
       condition: Expression,

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 Logging {
+abstract class LogicalPlan extends QueryPlan[LogicalPlan] with PredicateHelper with Logging {
 
   private var _analyzed: Boolean = false
 
@@ -127,33 +127,41 @@ abstract class LogicalPlan extends QueryPlan[LogicalPlan] with Logging {
       cleanLeft.children.size == cleanRight.children.size && {
       logDebug(
         s"[${cleanRight.cleanArgs.mkString(", ")}] == [${cleanLeft.cleanArgs.mkString(", ")}]")
-      cleanRight.cleanArgs == cleanLeft.cleanArgs
+      (cleanRight.cleanArgs == cleanLeft.cleanArgs) || ((cleanLeft, cleanRight) match {
+        case (l: Filter, r: Filter) =>
+          equivalentPredicates(cleanExpression(l.condition, l.children.flatMap(_.output)),
+            cleanExpression(r.condition, r.children.flatMap(_.output)))
+        case _ => false
+      })
     } &&
     (cleanLeft.children, cleanRight.children).zipped.forall(_ sameResult _)
   }
 
+  /** Clean an expression so that differences in expression id should not affect equality */
+  def cleanExpression(e: Expression, input: Seq[Attribute]): Expression = e match {
+    case a: Alias =>
+      // As the root of the expression, Alias will always take an arbitrary exprId, we need
+      // to erase that for equality testing.
+      val cleanedExprId = Alias(a.child, a.name)(ExprId(-1), a.qualifiers)
+      BindReferences.bindReference(cleanedExprId, input, allowFailures = true)
+    case other => BindReferences.bindReference(other, input, allowFailures = true)
+  }
+
+
   /** Args that have cleaned such that differences in expression id should not affect equality */
   protected lazy val cleanArgs: Seq[Any] = {
     val input = children.flatMap(_.output)
-    def cleanExpression(e: Expression) = e match {
-      case a: Alias =>
-        // As the root of the expression, Alias will always take an arbitrary exprId, we need
-        // to erase that for equality testing.
-        val cleanedExprId = Alias(a.child, a.name)(ExprId(-1), a.qualifiers)
-        BindReferences.bindReference(cleanedExprId, input, allowFailures = true)
-      case other => BindReferences.bindReference(other, input, allowFailures = true)
-    }
 
     productIterator.map {
       // Children are checked using sameResult above.
       case tn: TreeNode[_] if containsChild(tn) => null
-      case e: Expression => cleanExpression(e)
+      case e: Expression => cleanExpression(e, input)
       case s: Option[_] => s.map {
-        case e: Expression => cleanExpression(e)
+        case e: Expression => cleanExpression(e, input)
         case other => other
       }
       case s: Seq[_] => s.map {
-        case e: Expression => cleanExpression(e)
+        case e: Expression => cleanExpression(e, input)
         case other => other
       }
       case other => other

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

@@ -28,8 +28,8 @@ import org.apache.spark.sql.catalyst.util._
  * Tests for the sameResult function of [[LogicalPlan]].
  */
 class SameResultSuite extends SparkFunSuite {
-  val testRelation = LocalRelation('a.int, 'b.int, 'c.int)
-  val testRelation2 = LocalRelation('a.int, 'b.int, 'c.int)
+  val testRelation = LocalRelation('a.int, 'b.int, 'c.int, 'd.int)
+  val testRelation2 = LocalRelation('a.int, 'b.int, 'c.int, 'd.int)
 
   def assertSameResult(a: LogicalPlan, b: LogicalPlan, result: Boolean = true): Unit = {
     val aAnalyzed = a.analyze
@@ -57,6 +57,21 @@ class SameResultSuite extends SparkFunSuite {
 
   test("filters") {
     assertSameResult(testRelation.where('a === 'b), testRelation2.where('a === 'b))
+    assertSameResult(testRelation.where('a === 'b && 'c === 'd),
+      testRelation2.where('c === 'd && 'a === 'b )
+    )
+    assertSameResult(testRelation.where('a === 'b || 'c === 'd),
+      testRelation2.where('c === 'd || 'a === 'b )
+    )
+
+    assertSameResult(testRelation.where('a === 'b && 'c === 'd),
+      testRelation2.where('a === 'c && 'b === 'd),
+      result = false
+    )
+    assertSameResult(testRelation.where('a === 'b || 'c === 'd),
+      testRelation2.where('a === 'c || 'b === 'd),
+      result = false
+    )
   }
 
   test("sorts") {