Refactor to add null filter to joinConditions

it will be pushed down by other rules, such as PushPredicateThroughJoin

Author: vidmantas zemleris

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

@@ -986,29 +986,22 @@ object RemoveLiteralFromGroupExpressions extends Rule[LogicalPlan] {
  * For an inner join - remove rows with null keys on both sides
  */
 object JoinSkewOptimizer extends Rule[LogicalPlan] with PredicateHelper {
-  /**
-   * Adds a null filter on given columns, if any
-   */
-  def addNullFilter(columns: Seq[Expression], expr: LogicalPlan): LogicalPlan = {
-    columns.map(IsNotNull)
-      .reduceLeftOption(And)
-      .map(Filter(_, expr))
-      .getOrElse(expr)
-  }
-
   private def hasNullableKeys(leftKeys: Seq[Expression], rightKeys: Seq[Expression]) = {
     leftKeys.exists(_.nullable) || rightKeys.exists(_.nullable)
   }
 
   def apply(plan: LogicalPlan): LogicalPlan = plan transform {
     case join @ Join(left, right, joinType, originalJoinCondition) =>
       join match {
+        // add a non-null join-key filter on both sides of Inner or LeftSemi join
         case ExtractEquiJoinKeys(_, leftKeys, rightKeys, _, _, _)
-          if hasNullableKeys(leftKeys, rightKeys) && Seq(Inner, LeftSemi).contains(joinType) =>
-            // add a non-null join-key filter on both sides of join
-            val newLeft = addNullFilter(leftKeys.filter(_.nullable), left)
-            val newRight = addNullFilter(rightKeys.filter(_.nullable), right)
-            Join(newLeft, newRight, joinType, originalJoinCondition)
+          if Seq(Inner, LeftSemi).contains(joinType) && hasNullableKeys(leftKeys, rightKeys) =>
+            val nullFilters = (leftKeys ++ rightKeys)
+              .filter(_.nullable)
+              .map(IsNotNull)
+            val newJoinCondition = (originalJoinCondition ++ nullFilters).reduceLeftOption(And)
+
+            Join(left, right, joinType, newJoinCondition)
 
         case _ => join
       }

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

@@ -307,8 +307,8 @@ class FilterPushdownSuite extends PlanTest {
     }
 
     val optimized = Optimize.execute(originalQuery.analyze)
-    val left = testRelation.where('a.isNotNull).where('b >= 1)
-    val right = testRelation1.where('d.isNotNull).where('d >= 2)
+    val left = testRelation.where('b >= 1 && 'a.isNotNull)
+    val right = testRelation1.where('d >= 2 && 'd.isNotNull)
     val correctAnswer =
       left.join(right, LeftSemi, Option("a".attr === "d".attr)).analyze