extract non-deterministic expression in some case and add test case

Author: zhaorongsheng

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/patterns.scala

@@ -43,6 +43,8 @@ object PhysicalOperation extends PredicateHelper {
 
   /**
    * Collects all deterministic projects and filters, in-lining/substituting aliases if necessary.
+   * Note that, if the filter is non-deterministic the deterministic expressions should be
+   * extracted.
    * Here are two examples for alias in-lining/substitution.
    * Before:
    * {{{
@@ -63,9 +65,8 @@ object PhysicalOperation extends PredicateHelper {
         val substitutedFields = fields.map(substitute(aliases)).asInstanceOf[Seq[NamedExpression]]
         (Some(substitutedFields), filters, other, collectAliases(substitutedFields))
 
-      case filter @ Filter(condition, child)
-          if condition.deterministic || condition.isInstanceOf[And] =>
-        val determinedCondition = getDeterminedExpression(condition)
+      case filter @ Filter(condition, child) =>
+        val determinedCondition = extractDeterministicExpressions(condition)
         if (determinedCondition.isDefined) {
           val (fields, filters, other, aliases) = collectProjectsAndFilters(child)
           val substitutedCondition = substitute(aliases)(determinedCondition.get)
@@ -98,21 +99,48 @@ object PhysicalOperation extends PredicateHelper {
     }
   }
 
-  private def getDeterminedExpression(expr: Expression): Option[Expression] = {
+  /**
+   * Extract the deterministic expressions in non-deterministic expressions, i.e. 'And' and 'Or'.
+   *
+   * Example input:
+   * {{{
+   *   col = 1 and rand() < 1
+   *   (col1 = 1 and rand() < 1) and col2 = 1
+   *   col1 = 1 or rand() < 1
+   *   (col1 = 1 and rand() < 1) or (col2 = 1 and rand() < 1)
+   * }}}
+   *
+   * Result:
+   * {{{
+   *   col = 1
+   *   col1 = 1 and col2 = 1
+   *   None
+   *   col1 = 1 or col2 = 1
+   * }}}
+   */
+  private[planning] def extractDeterministicExpressions(expr: Expression): Option[Expression] = {
     if (expr.deterministic) {
       Some(expr)
     } else {
       expr match {
         case And(left, right) =>
-          val leftDeterminedExpr = getDeterminedExpression(left)
-          val rightDeterminedExpr = getDeterminedExpression(right)
+          val leftDeterminedExpr = extractDeterministicExpressions(left)
+          val rightDeterminedExpr = extractDeterministicExpressions(right)
           if (leftDeterminedExpr.isDefined && rightDeterminedExpr.isDefined) {
             Some(And(leftDeterminedExpr.get, rightDeterminedExpr.get))
           } else if (leftDeterminedExpr.isDefined || rightDeterminedExpr.isDefined) {
             Some(leftDeterminedExpr.getOrElse(rightDeterminedExpr.get))
           } else {
             None
           }
+        case Or(left, right) =>
+          val leftDeterminedExpr = extractDeterministicExpressions(left)
+          val rightDeterminedExpr = extractDeterministicExpressions(right)
+          if (leftDeterminedExpr.isDefined && rightDeterminedExpr.isDefined) {
+            Some(Or(leftDeterminedExpr.get, rightDeterminedExpr.get))
+          } else {
+            None
+          }
         case _ =>
           None
       }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/planning/PhysicalOperationSuite.scala

@@ -0,0 +1,45 @@
+/*
+ * 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.planning
+
+import org.apache.spark.sql.catalyst.dsl.expressions._
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.plans.PlanTest
+
+class PhysicalOperationSuite extends PlanTest {
+  import PhysicalOperation.extractDeterministicExpressions
+
+  test("Extract deterministic expressions") {
+    val expr1 = ('col.int === Literal(1)) && rand(1) < 1
+    val expr2 = 'col.int === Literal(1)
+    compareExpressions(extractDeterministicExpressions(expr1).get, expr2)
+
+    val expr3 = (('col1.int === Literal(1)) && rand(1) < 1) && ('col2.int === Literal(1))
+    val expr4 = ('col1.int === Literal(1)) && ('col2.int === Literal(1))
+    compareExpressions(extractDeterministicExpressions(expr3).get, expr4)
+
+    val expr5 = ('col.int === Literal(1)) || rand(1) < 1
+    val expr6 = Literal(null)
+    compareExpressions(extractDeterministicExpressions(expr5).getOrElse(Literal(null)), expr6)
+
+    val expr7 = (('col1.int === Literal(1)) && rand(1) < 1) ||
+        (('col2.int === Literal(1)) && rand(1) < 1)
+    val expr8 = ('col1.int === Literal(1)) || ('col2.int === Literal(1))
+    compareExpressions(extractDeterministicExpressions(expr7).getOrElse(Literal(null)), expr8)
+  }
+}