[SPARK-29213] Make it consistent when get notnull output and generate null checks in FilteExec

Author: wangshuo128

sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala

@@ -117,7 +117,7 @@ case class FilterExec(condition: Expression, child: SparkPlan)
 
   override def output: Seq[Attribute] = {
     child.output.map { a =>
-      if (a.nullable && notNullAttributes.contains(a.exprId)) {
+      if (a.nullable && notNullPreds.exists(_.semanticEquals(a))) {
         a.withNullability(false)
       } else {
         a

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala

@@ -2412,4 +2412,29 @@ class SQLQuerySuite extends QueryTest with SQLTestUtils with TestHiveSingleton {
       }
     }
   }
+
+  test("SPARK-29213 Make it consistent when get notnull output and generate null " +
+    "checks in FilterExec") {
+    withTable("table1", "table2", "table3") {
+      sql("create table table1(x string)")
+      sql("create table table2(x bigint)")
+      sql("create table table3(x string)")
+      sql("insert into table2 select null as x")
+      sql(
+        """
+          |select t1.x
+          |from (
+          |    select x from table1) t1
+          |left join (
+          |    select x from (
+          |        select x from table2
+          |        union all
+          |        select substr(x,5) x from table3
+          |    ) a
+          |    where length(x)>0
+          |) t3
+          |on t1.x=t3.x
+        """.stripMargin).collect()
+    }
+  }
 }