Avoid generating too many constraints

Author: wangyum

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

@@ -177,20 +177,13 @@ abstract class UnaryNode extends LogicalPlan {
       case a @ Alias(e, _) =>
         // For every alias in `projectList`, replace the reference in constraints by its attribute.
         allConstraints ++= allConstraints.map {
-          case binaryExpression @ EqualNullSafe(_: BinaryExpression, _) =>
-            binaryExpression
-          case ternaryExpression @ EqualNullSafe(_: TernaryExpression, _) =>
-            ternaryExpression
-          case quaternaryExpression @ EqualNullSafe(_: QuaternaryExpression, _) =>
-            quaternaryExpression
-          case septenaryExpression @ EqualNullSafe(_: SeptenaryExpression, _) =>
-            septenaryExpression
-          case userDefinedExpression @ EqualNullSafe(_: UserDefinedExpression, _) =>
-            userDefinedExpression
-          case other =>
-            other transform {
-              case expr: Expression if expr.semanticEquals(e) =>
-                a.toAttribute
+          case e @ EqualNullSafe(l, _: AttributeReference)
+            if !l.isInstanceOf[AttributeReference] => e
+          case e @ EqualNullSafe(_: AttributeReference, r)
+            if !r.isInstanceOf[AttributeReference] => e
+          case other => other transform {
+            case expr: Expression if expr.semanticEquals(e) =>
+              a.toAttribute
             }
         }
         allConstraints += EqualNullSafe(e, a.toAttribute)

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

@@ -422,4 +422,12 @@ class ConstraintPropagationSuite extends SparkFunSuite with PlanTest {
       assert(aliasedRelation.analyze.constraints.isEmpty)
     }
   }
+
+  test("SPARK-29606 Avoid generating too many constraints") {
+    val aliasedRelation = LocalRelation('a.int, 'b.int, 'c.int)
+      .select('a, 'b, 'c, ('a + 'b + 'c).as("abc"))
+      .select('a.as("a1"), 'b.as("b1"), 'c.as("c1"), 'abc.as("abc1"))
+    assert(aliasedRelation.analyze.isInstanceOf[Project])
+    assert(aliasedRelation.analyze.asInstanceOf[Project].validConstraints.size === 5)
+  }
 }