modify function inferAdditionalConstraints to avoid producing non-converging set of constraints

Author: jiangxb1987

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala

@@ -74,20 +74,36 @@ abstract class QueryPlan[PlanType <: QueryPlan[PlanType]] extends TreeNode[PlanT
    * additional constraint of the form `b = 5`
    */
   private def inferAdditionalConstraints(constraints: Set[Expression]): Set[Expression] = {
+    // Collect alias from expressions to avoid producing non-converging set of constraints
+    // for recursive functions.
+    // For more details, infer https://issues.apache.org/jira/browse/SPARK-17733
+    val aliasMap = AttributeMap((expressions ++ children.flatMap(_.expressions)).collect {
+      case a: Alias => (a.toAttribute, a.child)
+    })
+
     var inferredConstraints = Set.empty[Expression]
     constraints.foreach {
       case eq @ EqualTo(l: Attribute, r: Attribute) =>
         inferredConstraints ++= (constraints - eq).map(_ transform {
-          case a: Attribute if a.semanticEquals(l) => r
+          case a: Attribute if a.semanticEquals(l) && !isRecursiveDeduction(a, r, aliasMap) => r
         })
         inferredConstraints ++= (constraints - eq).map(_ transform {
-          case a: Attribute if a.semanticEquals(r) => l
+          case a: Attribute if a.semanticEquals(r) && !isRecursiveDeduction(l, a, aliasMap) => l
         })
       case _ => // No inference
     }
     inferredConstraints -- constraints
   }
 
+  private def isRecursiveDeduction(
+      left: Attribute,
+      right: Attribute,
+      aliasMap: AttributeMap[Expression]): Boolean = {
+    val leftExpression = aliasMap.getOrElse(left, left)
+    val rightExpression = aliasMap.getOrElse(right, right)
+    leftExpression.containsChild(rightExpression) || rightExpression.containsChild(leftExpression)
+  }
+
   /**
    * An [[ExpressionSet]] that contains invariants about the rows output by this operator. For
    * example, if this set contains the expression `a = 2` then that expression is guaranteed to

sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala

@@ -2678,4 +2678,45 @@ class SQLQuerySuite extends QueryTest with SharedSQLContext {
       }
     }
   }
+
+  test("SPARK-17733 InferFiltersFromConstraints rule never terminates for query") {
+    withTempView("tmpv") {
+      spark.range(10).toDF("a").createTempView("tmpv")
+
+      // Just ensure the following query will successfully execute complete.
+      assert(sql(
+        """
+          |SELECT
+          |  *
+          |FROM (
+          |  SELECT
+          |    COALESCE(t1.a, t2.a) AS int_col,
+          |    t1.a,
+          |    t2.a AS b
+          |  FROM tmpv t1
+          |  CROSS JOIN tmpv t2
+          |) t1
+          |INNER JOIN tmpv t2
+          |ON (((t2.a) = (t1.a)) AND ((t2.a) = (t1.int_col))) AND ((t2.a) = (t1.b))
+        """.stripMargin).count() > 0
+      )
+
+    //sql("CREATE TEMPORARY VIEW foo(a) AS VALUES (CAST(-993 AS BIGINT))")
+
+    /*sql(
+      """
+        |SELECT
+        |*
+        |FROM (
+        |    SELECT
+        |    COALESCE(t1.a, t2.a) AS int_col,
+        |    t1.a,
+        |    t2.a AS b
+        |    FROM foo t1
+        |    CROSS JOIN foo t2
+        |) t1
+        |INNER JOIN foo t2 ON (((t2.a) = (t1.a)) AND ((t2.a) = (t1.int_col))) AND ((t2.a) = (t1.b))
+      """.stripMargin).collect()*/
+    }
+  }
 }