Null value should be handled by NullPropagation.

Author: wangyum

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

@@ -472,6 +472,7 @@ object SimplifyConditionals extends Rule[LogicalPlan] with PredicateHelper {
 
   private def isAlwaysFalse(exps: Seq[Expression], equalTo: Literal): Boolean = {
     exps.forall(_.isInstanceOf[Literal]) &&
+      exps.forall(_.asInstanceOf[Literal].value != null) &&
       exps.forall(!EqualTo(_, equalTo).eval(EmptyRow).asInstanceOf[Boolean])
   }
 
@@ -529,12 +530,15 @@ object SimplifyConditionals extends Rule[LogicalPlan] with PredicateHelper {
           e.copy(branches = branches.take(i).map(branch => (branch._1, elseValue)))
         }
 
+      // Null value should be handled by NullPropagation.
       case EqualTo(i @ If(_, trueValue, falseValue), right: Literal)
-          if i.deterministic && isAlwaysFalse(trueValue :: falseValue :: Nil, right) =>
+          if i.deterministic && right.value != null &&
+            isAlwaysFalse(trueValue :: falseValue :: Nil, right) =>
         FalseLiteral
 
       case EqualTo(c @ CaseWhen(branches, elseValue), right: Literal)
-          if c.deterministic && isAlwaysFalse(branches.map(_._2) ++ elseValue, right) =>
+          if c.deterministic && right.value != null &&
+            isAlwaysFalse(branches.map(_._2) ++ elseValue, right) =>
         FalseLiteral
     }
   }

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

@@ -220,14 +220,13 @@ class SimplifyConditionalSuite extends PlanTest with ExpressionEvalHelper with P
     assert(!nonDeterministic.deterministic)
     assertEquivalent(EqualTo(nonDeterministic, Literal(-1)), EqualTo(nonDeterministic, Literal(-1)))
 
-    // null check, SPARK-33798 will change the following two behaviors.
+    // Null value should be handled by NullPropagation.
     assertEquivalent(
       EqualTo(If(a === Literal(1), Literal(null, IntegerType), Literal(1)), Literal(2)),
-      FalseLiteral)
+      EqualTo(If(a === Literal(1), Literal(null, IntegerType), Literal(1)), Literal(2)))
     assertEquivalent(
       EqualTo(If(a =!= Literal(1), Literal(1), Literal(2)), Literal(null, IntegerType)),
-      FalseLiteral)
-
+      EqualTo(If(a =!= Literal(1), Literal(1), Literal(2)), Literal(null, IntegerType)))
     assertEquivalent(
       EqualTo(If(a === Literal(1), Literal(null, IntegerType), Literal(1)), Literal(1)),
       EqualTo(If(a === Literal(1), Literal(null, IntegerType), Literal(1)), Literal(1)))
@@ -273,14 +272,13 @@ class SimplifyConditionalSuite extends PlanTest with ExpressionEvalHelper with P
     assert(!nonDeterministic.deterministic)
     assertEquivalent(EqualTo(nonDeterministic, Literal(-1)), EqualTo(nonDeterministic, Literal(-1)))
 
-    // null check, SPARK-33798 will change the following two behaviors.
+    // Null value should be handled by NullPropagation.
     assertEquivalent(
       EqualTo(CaseWhen(Seq((a, Literal(null, IntegerType))), Some(Literal(1))), Literal(2)),
-      FalseLiteral)
+      EqualTo(CaseWhen(Seq((a, Literal(null, IntegerType))), Some(Literal(1))), Literal(2)))
     assertEquivalent(
       EqualTo(CaseWhen(Seq((a, Literal(1))), Some(Literal(2))), Literal(null, IntegerType)),
-      FalseLiteral)
-
+      EqualTo(CaseWhen(Seq((a, Literal(1))), Some(Literal(2))), Literal(null, IntegerType)))
     assertEquivalent(
       EqualTo(CaseWhen(Seq((a, Literal(null, IntegerType))), Some(Literal(1))), Literal(1)),
       EqualTo(CaseWhen(Seq((a, Literal(null, IntegerType))), Some(Literal(1))), Literal(1)))