[SPARK-23079][SQL] Fix query constraints propagation with aliases

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

@@ -96,10 +96,18 @@ trait QueryPlanConstraints { self: LogicalPlan =>
 
   // Collect aliases from expressions of the whole tree rooted by the current QueryPlan node, so
   // we may avoid producing recursive constraints.
-  private lazy val aliasMap: AttributeMap[Expression] = AttributeMap(
-    expressions.collect {
+  private lazy val aliasMap: AttributeMap[Expression] = {
+    val childrenAliases = children.flatMap { child =>
+      val childOutputSet = child.outputSet
+      child.asInstanceOf[QueryPlanConstraints].aliasMap.filter {
+        case (_, c) => c.references.nonEmpty && c.references.subsetOf(childOutputSet)
+      }
+    }
+    AttributeMap(expressions.collect {
       case a: Alias if !a.child.isInstanceOf[Literal] => (a.toAttribute, a.child)
-    } ++ children.flatMap(_.asInstanceOf[QueryPlanConstraints].aliasMap))
+    } ++ childrenAliases)
+  }
+
     // Note: the explicit cast is necessary, since Scala compiler fails to infer the type.
 
   /**

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

@@ -2717,6 +2717,17 @@ class SQLQuerySuite extends QueryTest with SharedSQLContext {
     }
   }
 
+  test("SPARK-23079: constraints should be inferred correctly with aliases") {
+    withTable("t") {
+      spark.range(5).write.saveAsTable("t")
+      val t = spark.read.table("t")
+      val left = t.withColumn("xid", $"id" + lit(1)).as("x")
+      val right = t.withColumnRenamed("id", "xid").as("y")
+      val df = left.join(right, "xid").filter("id = 3").toDF()
+      checkAnswer(df, Row(4, 3))
+    }
+  }
+
   test("SRARK-22266: the same aggregate function was calculated multiple times") {
     val query = "SELECT a, max(b+1), max(b+1) + 1 FROM testData2 GROUP BY a"
     val df = sql(query)