[SPARK-29162][SQL] Simplify NOT(IsNull(x)) and NOT(IsNotNull(x))

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

@@ -373,6 +373,9 @@ object BooleanSimplification extends Rule[LogicalPlan] with PredicateHelper {
       case Not(a And b) => Or(Not(a), Not(b))
 
       case Not(Not(e)) => e
+
+      case Not(expr: IsNull) => IsNotNull(expr.child)
+      case Not(expr: IsNotNull) => IsNull(expr.child)
     }
   }
 }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/SimplifyConditionalSuite.scala

@@ -18,6 +18,7 @@
 package org.apache.spark.sql.catalyst.optimizer
 
 import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
+import org.apache.spark.sql.catalyst.dsl.expressions._
 import org.apache.spark.sql.catalyst.dsl.plans._
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.expressions.Literal.{FalseLiteral, TrueLiteral}
@@ -40,10 +41,19 @@ class SimplifyConditionalSuite extends PlanTest with PredicateHelper {
     comparePlans(actual, correctAnswer)
   }
 
+  private def assertFilter(originalExpr: Expression, expectedExpr: Expression): Unit = {
+    val originalPlan = testRelation.where(originalExpr).analyze
+    val optimizedPlan = Optimize.execute(originalPlan)
+    val expectedPlan = testRelation.where(expectedExpr).analyze
+    comparePlans(optimizedPlan, expectedPlan)
+  }
+
   private val trueBranch = (TrueLiteral, Literal(5))
   private val normalBranch = (NonFoldableLiteral(true), Literal(10))
   private val unreachableBranch = (FalseLiteral, Literal(20))
   private val nullBranch = (Literal.create(null, NullType), Literal(30))
+  private val testRelation =
+    LocalRelation('i.int, 'b.boolean, 'a.array(IntegerType), 'm.map(IntegerType, IntegerType))
 
   val isNotNullCond = IsNotNull(UnresolvedAttribute(Seq("a")))
   val isNullCond = IsNull(UnresolvedAttribute("b"))
@@ -166,4 +176,9 @@ class SimplifyConditionalSuite extends PlanTest with PredicateHelper {
         Literal(1))
     )
   }
+
+  test("simplify NOT(IsNull(x)) and NOT(IsNotNull(x))") {
+    assertFilter(Not(IsNotNull(UnresolvedAttribute("b"))), IsNull(UnresolvedAttribute("b")))
+    assertFilter(Not(IsNull(UnresolvedAttribute("b"))), IsNotNull(UnresolvedAttribute("b")))
+  }
 }

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

@@ -3192,6 +3192,19 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession {
       checkAnswer(df3, Array(Row(new java.math.BigDecimal("0.100000000000000000000000100"))))
     }
   }
+
+  test("SPARK-29152: Simplify NOT(IsNull(x)) and NOT(IsNotNull(x))") {
+    withTempView("tbl1") {
+      val df: DataFrame =
+        Seq[java.lang.Boolean](true, false, true, null, false, null, true).toDF("id")
+      df.createOrReplaceTempView("tbl1")
+      val query1 = sql("select id from tbl1 where not(isnull(id) or id == false)")
+      val query2 = sql("select id from tbl1 where not(isnotnull(id) and id == true)  ")
+
+      checkAnswer(query1, Row(true) :: Row(true) :: Row(true) :: Nil)
+      checkAnswer(query2, Row(false) :: Row(null) :: Row(false) :: Row(null) :: Nil)
+    }
+  }
 }
 
 case class Foo(bar: Option[String])