apache
diff --git a/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Expression.scala‎
Lines changed: 11 additions & 0 deletions b/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Expression.scala‎
Lines changed: 11 additions & 0 deletions
diff --git a/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala‎
Lines changed: 2 additions & 0 deletions b/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala‎
Lines changed: 2 additions & 0 deletions
diff --git a/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/ReplaceNullWithFalseInPredicate.scala‎
Lines changed: 9 additions & 5 deletions b/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/ReplaceNullWithFalseInPredicate.scala‎
Lines changed: 9 additions & 5 deletions
diff --git a/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/SimplifyConditionalsInPredicate.scala‎
Lines changed: 80 additions & 0 deletions b/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/SimplifyConditionalsInPredicate.scala‎
Lines changed: 80 additions & 0 deletions
diff --git a/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala‎
Lines changed: 99 additions & 3 deletions b/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala‎
Lines changed: 99 additions & 3 deletions
diff --git a/‎sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/BooleanSimplificationSuite.scala‎
Lines changed: 4 additions & 4 deletions b/‎sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/BooleanSimplificationSuite.scala‎
Lines changed: 4 additions & 4 deletions
@@ -521,6 +521,12 @@ abstract class UnaryExpression extends Expression {
   }
 }
 
+
+object UnaryExpression {
+  def unapply(e: UnaryExpression): Option[Expression] = Some(e.child)
+}
+
+
 /**
  * An expression with two inputs and one output. The output is by default evaluated to null
  * if any input is evaluated to null.
@@ -621,6 +627,11 @@ abstract class BinaryExpression extends Expression {
 }
 
 
+object BinaryExpression {
+  def unapply(e: BinaryExpression): Option[(Expression, Expression)] = Some((e.left, e.right))
+}
+
+
 /**
  * A [[BinaryExpression]] that is an operator, with two properties:
  *
 
@@ -96,9 +96,11 @@ abstract class Optimizer(catalogManager: CatalogManager)
         LikeSimplification,
         BooleanSimplification,
         SimplifyConditionals,
+        PushFoldableIntoBranches,
         RemoveDispensableExpressions,
         SimplifyBinaryComparison,
         ReplaceNullWithFalseInPredicate,
+        SimplifyConditionalsInPredicate,
         PruneFilters,
         SimplifyCasts,
         SimplifyCaseConversionExpressions,
 
@@ -17,12 +17,10 @@
 
 package org.apache.spark.sql.catalyst.optimizer
 
-import org.apache.spark.sql.catalyst.expressions.{And, ArrayExists, ArrayFilter, CaseWhen, Expression, If}
-import org.apache.spark.sql.catalyst.expressions.{LambdaFunction, Literal, MapFilter, Or}
-import org.apache.spark.sql.catalyst.expressions.Literal.FalseLiteral
+import org.apache.spark.sql.catalyst.expressions.{And, ArrayExists, ArrayFilter, CaseWhen, EqualNullSafe, Expression, If, LambdaFunction, Literal, MapFilter, Or}
+import org.apache.spark.sql.catalyst.expressions.Literal.{FalseLiteral, TrueLiteral}
 import org.apache.spark.sql.catalyst.plans.logical.{Filter, Join, LogicalPlan}
 import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.BooleanType
 import org.apache.spark.util.Utils
 
@@ -55,6 +53,12 @@ object ReplaceNullWithFalseInPredicate extends Rule[LogicalPlan] {
     case f @ Filter(cond, _) => f.copy(condition = replaceNullWithFalse(cond))
     case j @ Join(_, _, _, Some(cond), _) => j.copy(condition = Some(replaceNullWithFalse(cond)))
     case p: LogicalPlan => p transformExpressions {
+      // For `EqualNullSafe` with a `TrueLiteral`, whether the other side is null or false has no
+      // difference, as `null <=> true` and `false <=> true` both return false.
+      case EqualNullSafe(left, TrueLiteral) =>
+        EqualNullSafe(replaceNullWithFalse(left), TrueLiteral)
+      case EqualNullSafe(TrueLiteral, right) =>
+        EqualNullSafe(TrueLiteral, replaceNullWithFalse(right))
       case i @ If(pred, _, _) => i.copy(predicate = replaceNullWithFalse(pred))
       case cw @ CaseWhen(branches, _) =>
         val newBranches = branches.map { case (cond, value) =>
@@ -92,7 +96,7 @@ object ReplaceNullWithFalseInPredicate extends Rule[LogicalPlan] {
       val newBranches = cw.branches.map { case (cond, value) =>
         replaceNullWithFalse(cond) -> replaceNullWithFalse(value)
       }
-      val newElseValue = cw.elseValue.map(replaceNullWithFalse)
+      val newElseValue = cw.elseValue.map(replaceNullWithFalse).getOrElse(FalseLiteral)
       CaseWhen(newBranches, newElseValue)
     case i @ If(pred, trueVal, falseVal) if i.dataType == BooleanType =>
       If(replaceNullWithFalse(pred), replaceNullWithFalse(trueVal), replaceNullWithFalse(falseVal))
 
@@ -0,0 +1,80 @@
+/*
+ * 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.expressions.{And, CaseWhen, Expression, If, Literal, Not, Or}
+import org.apache.spark.sql.catalyst.expressions.Literal.{FalseLiteral, TrueLiteral}
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.types.BooleanType
+
+/**
+ * A rule that converts conditional expressions to predicate expressions, if possible, in the
+ * search condition of the WHERE/HAVING/ON(JOIN) clauses, which contain an implicit Boolean operator
+ * "(search condition) = TRUE". After this converting, we can potentially push the filter down to
+ * the data source.
+ *
+ * Supported cases are:
+ * - IF(cond, trueVal, false)                   => AND(cond, trueVal)
+ * - IF(cond, trueVal, true)                    => OR(NOT(cond), trueVal)
+ * - IF(cond, false, falseVal)                  => AND(NOT(cond), elseVal)
+ * - IF(cond, true, falseVal)                   => OR(cond, elseVal)
+ * - CASE WHEN cond THEN trueVal ELSE false END => AND(cond, trueVal)
+ * - CASE WHEN cond THEN trueVal END            => AND(cond, trueVal)
+ * - CASE WHEN cond THEN trueVal ELSE null END  => AND(cond, trueVal)
+ * - CASE WHEN cond THEN trueVal ELSE true END  => OR(NOT(cond), trueVal)
+ * - CASE WHEN cond THEN false ELSE elseVal END => AND(NOT(cond), elseVal)
+ * - CASE WHEN cond THEN false END              => false
+ * - CASE WHEN cond THEN true ELSE elseVal END  => OR(cond, elseVal)
+ * - CASE WHEN cond THEN true END               => cond
+ */
+object SimplifyConditionalsInPredicate extends Rule[LogicalPlan] {
+
+  def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+    case f @ Filter(cond, _) => f.copy(condition = simplifyConditional(cond))
+    case j @ Join(_, _, _, Some(cond), _) => j.copy(condition = Some(simplifyConditional(cond)))
+  }
+
+  private def simplifyConditional(e: Expression): Expression = e match {
+    case And(left, right) => And(simplifyConditional(left), simplifyConditional(right))
+    case Or(left, right) => Or(simplifyConditional(left), simplifyConditional(right))
+    case If(cond, trueValue, FalseLiteral) => And(cond, trueValue)
+    case If(cond, trueValue, TrueLiteral) => Or(Not(cond), trueValue)
+    case If(cond, FalseLiteral, falseValue) => And(Not(cond), falseValue)
+    case If(cond, TrueLiteral, falseValue) => Or(cond, falseValue)
+    case CaseWhen(Seq((cond, trueValue)),
+        Some(FalseLiteral) | Some(Literal(null, BooleanType)) | None) =>
+      And(cond, trueValue)
+    case CaseWhen(Seq((cond, trueValue)), Some(TrueLiteral)) =>
+      Or(Not(cond), trueValue)
+    case CaseWhen(Seq((_, FalseLiteral)), Some(FalseLiteral) | None) =>
+      FalseLiteral
+    case CaseWhen(Seq((cond, FalseLiteral)), Some(elseValue)) =>
+      And(Not(cond), elseValue)
+    case CaseWhen(Seq((cond, TrueLiteral)), Some(FalseLiteral) | None) =>
+      cond
+    case CaseWhen(Seq((cond, TrueLiteral)), Some(elseValue)) =>
+      Or(cond, elseValue)
+    case e if e.dataType == BooleanType => e
+    case e =>
+      assert(e.dataType != BooleanType,
+      "Expected a Boolean type expression in SimplifyConditionalsInPredicate, " +
+        s"but got the type `${e.dataType.catalogString}` in `${e.sql}`.")
+      e
+  }
+}
@@ -21,7 +21,7 @@ import scala.collection.immutable.HashSet
 import scala.collection.mutable.{ArrayBuffer, Stack}
 
 import org.apache.spark.sql.catalyst.analysis._
-import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.{BinaryExpression, _}
 import org.apache.spark.sql.catalyst.expressions.Literal.{FalseLiteral, TrueLiteral}
 import org.apache.spark.sql.catalyst.expressions.aggregate._
 import org.apache.spark.sql.catalyst.expressions.objects.AssertNotNull
@@ -461,8 +461,21 @@ object SimplifyConditionals extends Rule[LogicalPlan] with PredicateHelper {
       case If(TrueLiteral, trueValue, _) => trueValue
       case If(FalseLiteral, _, falseValue) => falseValue
       case If(Literal(null, _), _, falseValue) => falseValue
+      case If(cond, TrueLiteral, FalseLiteral) =>
+        if (cond.nullable) EqualNullSafe(cond, TrueLiteral) else cond
+      case If(cond, FalseLiteral, TrueLiteral) =>
+        if (cond.nullable) Not(EqualNullSafe(cond, TrueLiteral)) else Not(cond)
       case If(cond, trueValue, falseValue)
         if cond.deterministic && trueValue.semanticEquals(falseValue) => trueValue
+      case If(cond, l @ Literal(null, _), FalseLiteral) if !cond.nullable => And(cond, l)
+      case If(cond, l @ Literal(null, _), TrueLiteral) if !cond.nullable => Or(Not(cond), l)
+      case If(cond, FalseLiteral, l @ Literal(null, _)) if !cond.nullable => And(Not(cond), l)
+      case If(cond, TrueLiteral, l @ Literal(null, _)) if !cond.nullable => Or(cond, l)
+
+      case CaseWhen(Seq((cond, TrueLiteral)), Some(FalseLiteral)) =>
+        if (cond.nullable) EqualNullSafe(cond, TrueLiteral) else cond
+      case CaseWhen(Seq((cond, FalseLiteral)), Some(TrueLiteral)) =>
+        if (cond.nullable) Not(EqualNullSafe(cond, TrueLiteral)) else Not(cond)
 
       case e @ CaseWhen(branches, elseValue) if branches.exists(x => falseOrNullLiteral(x._1)) =>
         // If there are branches that are always false, remove them.
@@ -488,8 +501,9 @@ object SimplifyConditionals extends Rule[LogicalPlan] with PredicateHelper {
         val (h, t) = branches.span(_._1 != TrueLiteral)
         CaseWhen( h :+ t.head, None)
 
-      case e @ CaseWhen(branches, Some(elseValue))
-          if branches.forall(_._2.semanticEquals(elseValue)) =>
+      case e @ CaseWhen(branches, elseOpt)
+          if branches.forall(_._2.semanticEquals(elseOpt.getOrElse(Literal(null, e.dataType)))) =>
+        val elseValue = elseOpt.getOrElse(Literal(null, e.dataType))
         // For non-deterministic conditions with side effect, we can not remove it, or change
         // the ordering. As a result, we try to remove the deterministic conditions from the tail.
         var hitNonDeterministicCond = false
@@ -510,6 +524,88 @@ object SimplifyConditionals extends Rule[LogicalPlan] with PredicateHelper {
 }
 
 
+/**
+ * Push the foldable expression into (if / case) branches.
+ */
+object PushFoldableIntoBranches extends Rule[LogicalPlan] with PredicateHelper {
+
+  // To be conservative here: it's only a guaranteed win if all but at most only one branch
+  // end up being not foldable.
+  private def atMostOneUnfoldable(exprs: Seq[Expression]): Boolean = {
+    val (foldables, others) = exprs.partition(_.foldable)
+    foldables.nonEmpty && others.length < 2
+  }
+
+  // Not all UnaryExpression can be pushed into (if / case) branches, e.g. Alias.
+  private def supportedUnaryExpression(e: UnaryExpression): Boolean = e match {
+    case _: IsNull | _: IsNotNull => true
+    case _: UnaryMathExpression | _: Abs | _: Bin | _: Factorial | _: Hex => true
+    case _: String2StringExpression | _: Ascii | _: Base64 | _: BitLength | _: Chr | _: Length =>
+      true
+    case _: CastBase => true
+    case _: LastDay => true
+    case _: ExtractIntervalPart => true
+    case _: ArraySetLike => true
+    case _: ExtractValue => true
+    case _ => false
+  }
+
+  // Not all BinaryExpression can be pushed into (if / case) branches.
+  private def supportedBinaryExpression(e: BinaryExpression): Boolean = e match {
+    case _: BinaryComparison | _: StringPredicate | _: StringRegexExpression => true
+    case _: BinaryArithmetic => true
+    case _: BinaryMathExpression => true
+    case _: AddMonths | _: DateAdd | _: DateAddInterval | _: DateDiff | _: DateSub => true
+    case _: FindInSet | _: RoundBase => true
+    case _ => false
+  }
+
+  def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+    case q: LogicalPlan => q transformExpressionsUp {
+      case u @ UnaryExpression(i @ If(_, trueValue, falseValue))
+          if supportedUnaryExpression(u) && atMostOneUnfoldable(Seq(trueValue, falseValue)) =>
+        i.copy(
+          trueValue = u.withNewChildren(Array(trueValue)),
+          falseValue = u.withNewChildren(Array(falseValue)))
+
+      case u @ UnaryExpression(c @ CaseWhen(branches, elseValue))
+          if supportedUnaryExpression(u) && atMostOneUnfoldable(branches.map(_._2) ++ elseValue) =>
+        c.copy(
+          branches.map(e => e.copy(_2 = u.withNewChildren(Array(e._2)))),
+          elseValue.map(e => u.withNewChildren(Array(e))))
+
+      case b @ BinaryExpression(i @ If(_, trueValue, falseValue), right)
+          if supportedBinaryExpression(b) && right.foldable &&
+            atMostOneUnfoldable(Seq(trueValue, falseValue)) =>
+        i.copy(
+          trueValue = b.withNewChildren(Array(trueValue, right)),
+          falseValue = b.withNewChildren(Array(falseValue, right)))
+
+      case b @ BinaryExpression(left, i @ If(_, trueValue, falseValue))
+          if supportedBinaryExpression(b) && left.foldable &&
+            atMostOneUnfoldable(Seq(trueValue, falseValue)) =>
+        i.copy(
+          trueValue = b.withNewChildren(Array(left, trueValue)),
+          falseValue = b.withNewChildren(Array(left, falseValue)))
+
+      case b @ BinaryExpression(c @ CaseWhen(branches, elseValue), right)
+          if supportedBinaryExpression(b) && right.foldable &&
+            atMostOneUnfoldable(branches.map(_._2) ++ elseValue) =>
+        c.copy(
+          branches.map(e => e.copy(_2 = b.withNewChildren(Array(e._2, right)))),
+          elseValue.map(e => b.withNewChildren(Array(e, right))))
+
+      case b @ BinaryExpression(left, c @ CaseWhen(branches, elseValue))
+          if supportedBinaryExpression(b) && left.foldable &&
+            atMostOneUnfoldable(branches.map(_._2) ++ elseValue) =>
+        c.copy(
+          branches.map(e => e.copy(_2 = b.withNewChildren(Array(left, e._2)))),
+          elseValue.map(e => b.withNewChildren(Array(left, e))))
+    }
+  }
+}
+
+
 /**
  * Simplifies LIKE expressions that do not need full regular expressions to evaluate the condition.
  * For example, when the expression is just checking to see if a string starts with a given
 
@@ -218,14 +218,14 @@ class BooleanSimplificationSuite extends PlanTest with ExpressionEvalHelper with
 
   test("Complementation Laws - null handling") {
     checkCondition('e && !'e,
-      testRelationWithData.where(If('e.isNull, Literal.create(null, BooleanType), false)).analyze)
+      testRelationWithData.where(And(Literal(null, BooleanType), 'e.isNull)).analyze)
     checkCondition(!'e && 'e,
-      testRelationWithData.where(If('e.isNull, Literal.create(null, BooleanType), false)).analyze)
+      testRelationWithData.where(And(Literal(null, BooleanType), 'e.isNull)).analyze)
 
     checkCondition('e || !'e,
-      testRelationWithData.where(If('e.isNull, Literal.create(null, BooleanType), true)).analyze)
+      testRelationWithData.where(Or('e.isNotNull, Literal(null, BooleanType))).analyze)
     checkCondition(!'e || 'e,
-      testRelationWithData.where(If('e.isNull, Literal.create(null, BooleanType), true)).analyze)
+      testRelationWithData.where(Or('e.isNotNull, Literal(null, BooleanType))).analyze)
   }
 
   test("Complementation Laws - negative case") {
Original file line number	Diff line number	Diff line change
`@@ -521,6 +521,12 @@ abstract class UnaryExpression extends Expression {`
`521`	`521`	`}`
`522`	`522`	`}`
`523`	`523`
	`524`	`+`
	`525`	`+object UnaryExpression {`
	`526`	`+ def unapply(e: UnaryExpression): Option[Expression] = Some(e.child)`
	`527`	`+}`
	`528`	`+`
	`529`	`+`
`524`	`530`	`/**`
`525`	`531`	`* An expression with two inputs and one output. The output is by default evaluated to null`
`526`	`532`	`* if any input is evaluated to null.`
`@@ -621,6 +627,11 @@ abstract class BinaryExpression extends Expression {`
`621`	`627`	`}`
`622`	`628`
`623`	`629`
	`630`	`+object BinaryExpression {`
	`631`	`+ def unapply(e: BinaryExpression): Option[(Expression, Expression)] = Some((e.left, e.right))`
	`632`	`+}`
	`633`	`+`
	`634`	`+`
`624`	`635`	`/**`
`625`	`636`	`* A [[BinaryExpression]] that is an operator, with two properties:`
`626`	`637`	`*`