[SPARK-32721][SQL] Simplify if clauses with null and boolean

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

@@ -103,6 +103,7 @@ abstract class Optimizer(catalogManager: CatalogManager)
         ReplaceNullWithFalseInPredicate,
         PruneFilters,
         SimplifyCasts,
+        SimplifyIf,
         SimplifyCaseConversionExpressions,
         RewriteCorrelatedScalarSubquery,
         EliminateSerialization,

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

@@ -716,6 +716,17 @@ object SimplifyCasts extends Rule[LogicalPlan] {
   }
 }
 
+/**
+ * Simplify if clauses of pattern `if(p, null, true|false)`, by replacing them with
+ * AND or OR clauses which are simpler and can better be pushed down
+ */
+object SimplifyIf extends Rule[LogicalPlan] {
+  val nullLiteral = Literal(null, BooleanType)
+  def apply(plan: LogicalPlan): LogicalPlan = plan transformAllExpressions {
+    case If(p, Literal(null, _), FalseLiteral) => And(p, nullLiteral)
+    case If(p, Literal(null, _), TrueLiteral) => Or(p, nullLiteral)
+  }
+}
 
 /**
  * Removes nodes that are not necessary.

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

@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.dsl.expressions._
+import org.apache.spark.sql.catalyst.dsl.plans.DslLogicalPlan
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.PlanTest
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.rules.RuleExecutor
+import org.apache.spark.sql.types.BooleanType
+
+class SimplifyIfSuite extends PlanTest {
+
+  object Optimize extends RuleExecutor[LogicalPlan] {
+    val batches = Batch("SimplifyIf", FixedPoint(1), SimplifyIf) :: Nil
+  }
+
+  val testRelation = LocalRelation('a.int)
+
+  test("simplify if when null is on the then branch") {
+    val nullLiteral = Literal(null, BooleanType)
+    assertEquivalent(If('a > 42, nullLiteral, false), And('a > 42, nullLiteral))
+    assertEquivalent(If('a > 42, nullLiteral, true), Or('a > 42, nullLiteral))
+  }
+
+  private def assertEquivalent(e1: Expression, e2: Expression) = {
+    val plan = testRelation.where(e1).analyze
+    val actual = Optimize.execute(plan)
+    val expected = testRelation.where(e2).analyze
+    comparePlans(actual, expected)
+  }
+
+}