fix

Author: wangyum

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

@@ -525,6 +525,9 @@ object SimplifyConditionals extends Rule[LogicalPlan] with PredicateHelper {
         } else {
           e.copy(branches = branches.take(i).map(branch => (branch._1, elseValue)))
         }
+
+      case e @ CaseWhen(_, elseValue) if elseValue.isEmpty =>
+        e.copy(elseValue = Some(Literal.create(null, e.dataType)))
     }
   }
 }

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

@@ -122,19 +122,20 @@ class PushFoldableIntoBranchesSuite
       CaseWhen(Seq((a, FalseLiteral), (c, FalseLiteral)), Some(TrueLiteral)))
     assertEquivalent(
       EqualTo(CaseWhen(Seq((a, Literal(1)), (c, Literal(2))), None), Literal(4)),
-      CaseWhen(Seq((a, FalseLiteral), (c, FalseLiteral)), None))
+      CaseWhen(Seq((a, FalseLiteral), (c, FalseLiteral)), Literal.create(null, BooleanType)))
 
     assertEquivalent(
       And(EqualTo(caseWhen, Literal(5)), EqualTo(caseWhen, Literal(6))),
       FalseLiteral)
 
     // Push down at most one branch is not foldable expressions.
     assertEquivalent(EqualTo(CaseWhen(Seq((a, b), (c, Literal(1))), None), Literal(1)),
-      CaseWhen(Seq((a, EqualTo(b, Literal(1))), (c, TrueLiteral)), None))
+      CaseWhen(Seq((a, EqualTo(b, Literal(1))), (c, TrueLiteral)),
+        Literal.create(null, BooleanType)))
     assertEquivalent(EqualTo(CaseWhen(Seq((a, b), (c, b + 1)), None), Literal(1)),
-      EqualTo(CaseWhen(Seq((a, b), (c, b + 1)), None), Literal(1)))
+      EqualTo(CaseWhen(Seq((a, b), (c, b + 1)), Literal.create(null, IntegerType)), Literal(1)))
     assertEquivalent(EqualTo(CaseWhen(Seq((a, b)), None), Literal(1)),
-      EqualTo(CaseWhen(Seq((a, b)), None), Literal(1)))
+      CaseWhen(Seq((a, b === Literal(1))), Literal.create(null, BooleanType)))
 
     // Push down non-deterministic expressions.
     val nonDeterministic =

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

@@ -115,7 +115,7 @@ class ReplaceNullWithFalseInPredicateSuite extends PlanTest {
     val expectedBranches = Seq(
       (UnresolvedAttribute("i") < Literal(10)) -> FalseLiteral,
       (UnresolvedAttribute("i") > Literal(40)) -> TrueLiteral)
-    val expectedCond = CaseWhen(expectedBranches)
+    val expectedCond = CaseWhen(expectedBranches, FalseLiteral)
 
     testFilter(originalCond, expectedCond)
     testJoin(originalCond, expectedCond)
@@ -246,7 +246,8 @@ class ReplaceNullWithFalseInPredicateSuite extends PlanTest {
     val expectedCond = CaseWhen(Seq((UnresolvedAttribute("i") > Literal(10)) ->
       CaseWhen(
         Seq((UnresolvedAttribute("i") > Literal(20)) -> TrueLiteral),
-        FalseLiteral)))
+        FalseLiteral)),
+      FalseLiteral)
     testFilter(originalCond = condition, expectedCond = expectedCond)
     testJoin(originalCond = condition, expectedCond = expectedCond)
     testDelete(originalCond = condition, expectedCond = expectedCond)
@@ -394,7 +395,7 @@ class ReplaceNullWithFalseInPredicateSuite extends PlanTest {
     val nonAllFalseBranches = Seq(
       (UnresolvedAttribute("i") < Literal(10)) -> FalseLiteral,
       (UnresolvedAttribute("i") > Literal(40)) -> TrueLiteral)
-    val nonAllFalseCond = CaseWhen(nonAllFalseBranches)
+    val nonAllFalseCond = CaseWhen(nonAllFalseBranches, FalseLiteral)
 
     testFilter(allFalseCond, FalseLiteral)
     testJoin(allFalseCond, FalseLiteral)

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

@@ -83,7 +83,7 @@ class SimplifyConditionalSuite extends PlanTest with ExpressionEvalHelper with P
     // i.e. removing branches whose conditions are always false
     assertEquivalent(
       CaseWhen(unreachableBranch :: normalBranch :: unreachableBranch :: nullBranch :: Nil, None),
-      CaseWhen(normalBranch :: Nil, None))
+      CaseWhen(normalBranch :: Nil, Literal.create(null, IntegerType)))
   }
 
   test("remove entire CaseWhen if only the else branch is reachable") {
@@ -215,4 +215,10 @@ class SimplifyConditionalSuite extends PlanTest with ExpressionEvalHelper with P
       If(GreaterThan(Rand(0), UnresolvedAttribute("a")), FalseLiteral, TrueLiteral),
       LessThanOrEqual(Rand(0), UnresolvedAttribute("a")))
   }
+
+  test("SPARK-33847: Replace None of elseValue inside CaseWhen to null literal") {
+    assertEquivalent(
+      CaseWhen(normalBranch :: Nil, None),
+      CaseWhen(normalBranch :: Nil, Literal.create(null, IntegerType)))
+  }
 }