This rule should be executed after ReplaceNullWithFalseInPredicate

Author: wangyum

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

@@ -146,7 +146,9 @@ abstract class Optimizer(catalogManager: CatalogManager)
         operatorOptimizationRuleSet: _*) ::
       Batch("Push extra predicate through join", fixedPoint,
         PushExtraPredicateThroughJoin,
-        PushDownPredicates) :: Nil
+        PushDownPredicates) ::
+      Batch("Pull out complex join condition", Once,
+        PullOutComplexJoinCondition) :: Nil
     }
 
     val batches = (
@@ -289,7 +291,6 @@ abstract class Optimizer(catalogManager: CatalogManager)
       ReplaceExpressions,
       RewriteNonCorrelatedExists,
       PullOutGroupingExpressions,
-      PullOutComplexJoinCondition,
       ComputeCurrentTime,
       ReplaceCurrentLike(catalogManager),
       SpecialDatetimeValues,

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

@@ -39,10 +39,12 @@ import org.apache.spark.sql.catalyst.trees.TreePattern.JOIN
  *      +- Filter isnotnull(x#2)
  *         +- Relation default.t2[x#2,y#3] parquet
  * }}}
+ *
+ *  This rule should be executed after ReplaceNullWithFalseInPredicate.
  */
 object PullOutComplexJoinCondition extends Rule[LogicalPlan] with PredicateHelper {
   def apply(plan: LogicalPlan): LogicalPlan = plan.transformUpWithPruning(_.containsPattern(JOIN)) {
-    case j @ Join(left, right, _, Some(condition), _) if j.resolved =>
+    case j @ Join(left, right, _, Some(condition), _) if !j.isStreaming =>
       val complexExps = splitConjunctivePredicates(condition).flatMap {
         case p: Expression => p.children.filter(e => !e.foldable && e.children.nonEmpty)
         case _ => Nil