fix

Author: wangyum

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

@@ -93,9 +93,12 @@ object ReplaceNullWithFalseInPredicate extends Rule[LogicalPlan] {
       val newBranches = cw.branches.map { case (cond, value) =>
         replaceNullWithFalse(cond) -> replaceNullWithFalse(value)
       }
-      val newElseValue = cw.elseValue.map(replaceNullWithFalse)
-        .orElse(if (newBranches.forall(_._2 == FalseLiteral)) Some(FalseLiteral) else None)
-      CaseWhen(newBranches, newElseValue)
+      if (newBranches.forall(_._2 == FalseLiteral) && cw.elseValue.isEmpty) {
+        FalseLiteral
+      } else {
+        val newElseValue = cw.elseValue.map(replaceNullWithFalse)
+        CaseWhen(newBranches, newElseValue)
+      }
     case i @ If(pred, trueVal, falseVal) if i.dataType == BooleanType =>
       If(replaceNullWithFalse(pred), replaceNullWithFalse(trueVal), replaceNullWithFalse(falseVal))
     case e if e.dataType == BooleanType =>

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

@@ -526,8 +526,8 @@ object SimplifyConditionals extends Rule[LogicalPlan] with PredicateHelper {
           e.copy(branches = branches.take(i).map(branch => (branch._1, elseValue)))
         }
 
-      case e @ CaseWhen(branches, elseOpt)
-          if elseOpt.isEmpty && branches.forall(_._2.semanticEquals(Literal(null, e.dataType))) =>
+      case e @ CaseWhen(branches, None)
+          if branches.forall(_._2.semanticEquals(Literal(null, e.dataType))) =>
         Literal(null, e.dataType)
     }
   }

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

@@ -258,4 +258,22 @@ class PushFoldableIntoBranchesSuite
       EqualTo(CaseWhen(Seq((a, Literal(1)), (c, Literal(2))), None).cast(StringType), Literal("4")),
       CaseWhen(Seq((a, FalseLiteral), (c, FalseLiteral)), None))
   }
+
+  test("SPARK-33847: Remove the CaseWhen if elseValue is empty and other outputs are null") {
+    assertEquivalent(
+      EqualTo(CaseWhen(Seq((a, Literal.create(null, IntegerType)))), Literal(2)),
+      Literal.create(null, BooleanType))
+    assertEquivalent(
+      EqualTo(CaseWhen(Seq((LessThan(Rand(1), Literal(0.5)), Literal.create(null, IntegerType)))),
+        Literal(2)),
+      Literal.create(null, BooleanType))
+
+    assertEquivalent(
+      EqualTo(CaseWhen(Seq((a, Literal("str")))).cast(IntegerType), Literal(2)),
+      Literal.create(null, BooleanType))
+    assertEquivalent(
+      EqualTo(CaseWhen(Seq((LessThan(Rand(1), Literal(0.5)), Literal("str")))).cast(IntegerType),
+        Literal(2)),
+      Literal.create(null, BooleanType))
+  }
 }