Code refine

Author: maryannxue

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

@@ -679,26 +679,26 @@ object InferFiltersFromConstraints extends Rule[LogicalPlan] with PredicateHelpe
       // Infer filter for left/right outer joins
       val newLeftOpt = joinType match {
         case RightOuter if newConditionOpt.isDefined =>
-          val rightConstraints = right.constraints.union(
-            splitConjunctivePredicates(newConditionOpt.get).toSet)
-          val inferredConstraints = ExpressionSet(
-            QueryPlanConstraints.inferAdditionalConstraints(rightConstraints))
-          val leftConditions = inferredConstraints
-            .filter(_.deterministic)
-            .filter(_.references.subsetOf(left.outputSet))
-          leftConditions.reduceLeftOption(And).map(Filter(_, left))
+          val inferredConstraints = left.getRelevantConstraints(
+            left.constraints
+              .union(right.constraints)
+              .union(splitConjunctivePredicates(newConditionOpt.get).toSet))
+          val newFilters = inferredConstraints
+            .filterNot(left.constraints.contains)
+            .reduceLeftOption(And)
+          newFilters.map(Filter(_, left))
         case _ => None
       }
       val newRightOpt = joinType match {
         case LeftOuter if newConditionOpt.isDefined =>
-          val leftConstraints = left.constraints.union(
-            splitConjunctivePredicates(newConditionOpt.get).toSet)
-          val inferredConstraints = ExpressionSet(
-            QueryPlanConstraints.inferAdditionalConstraints(leftConstraints))
-          val rightConditions = inferredConstraints
-            .filter(_.deterministic)
-            .filter(_.references.subsetOf(right.outputSet))
-          rightConditions.reduceLeftOption(And).map(Filter(_, right))
+          val inferredConstraints = right.getRelevantConstraints(
+            right.constraints
+              .union(left.constraints)
+              .union(splitConjunctivePredicates(newConditionOpt.get).toSet))
+          val newFilters = inferredConstraints
+            .filterNot(right.constraints.contains)
+            .reduceLeftOption(And)
+          newFilters.map(Filter(_, right))
         case _ => None
       }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala

@@ -29,7 +29,7 @@ trait QueryPlanConstraints { self: LogicalPlan =>
   lazy val allConstraints: ExpressionSet = {
     if (conf.constraintPropagationEnabled) {
       ExpressionSet(validConstraints
-        .union(QueryPlanConstraints.inferAdditionalConstraints(validConstraints))
+        .union(inferAdditionalConstraints(validConstraints))
         .union(constructIsNotNullConstraints(validConstraints)))
     } else {
       ExpressionSet(Set.empty)
@@ -41,9 +41,7 @@ trait QueryPlanConstraints { self: LogicalPlan =>
    * example, if this set contains the expression `a = 2` then that expression is guaranteed to
    * evaluate to `true` for all rows produced.
    */
-  lazy val constraints: ExpressionSet = ExpressionSet(allConstraints.filter { c =>
-        c.references.nonEmpty && c.references.subsetOf(outputSet) && c.deterministic
-      })
+  lazy val constraints: ExpressionSet = ExpressionSet(allConstraints.filter(selfReferenceOnly))
 
   /**
    * This method can be overridden by any child class of QueryPlan to specify a set of constraints
@@ -55,6 +53,23 @@ trait QueryPlanConstraints { self: LogicalPlan =>
    */
   protected def validConstraints: Set[Expression] = Set.empty
 
+  /**
+   * Returns an [[ExpressionSet]] that contains an additional set of constraints, such as
+   * equality constraints and `isNotNull` constraints, etc., and that only contains references
+   * to this [[LogicalPlan]] node.
+   */
+  def getRelevantConstraints(constraints: Set[Expression]): ExpressionSet = {
+    val allRelevantConstraints =
+      if (conf.constraintPropagationEnabled) {
+        constraints
+          .union(inferAdditionalConstraints(constraints))
+          .union(constructIsNotNullConstraints(constraints))
+      } else {
+        constraints
+      }
+    ExpressionSet(allRelevantConstraints.filter(selfReferenceOnly))
+  }
+
   /**
    * Infers a set of `isNotNull` constraints from null intolerant expressions as well as
    * non-nullable attributes. For e.g., if an expression is of the form (`a > 5`), this
@@ -96,16 +111,13 @@ trait QueryPlanConstraints { self: LogicalPlan =>
     case _: NullIntolerant => expr.children.flatMap(scanNullIntolerantAttribute)
     case _ => Seq.empty[Attribute]
   }
-}
-
-object QueryPlanConstraints {
 
   /**
    * Infers an additional set of constraints from a given set of equality constraints.
    * For e.g., if an operator has constraints of the form (`a = 5`, `a = b`), this returns an
    * additional constraint of the form `b = 5`.
    */
-  def inferAdditionalConstraints(constraints: Set[Expression]): Set[Expression] = {
+  private def inferAdditionalConstraints(constraints: Set[Expression]): Set[Expression] = {
     var inferredConstraints = Set.empty[Expression]
     constraints.foreach {
       case eq @ EqualTo(l: Attribute, r: Attribute) =>
@@ -123,4 +135,8 @@ object QueryPlanConstraints {
       destination: Attribute): Set[Expression] = constraints.map(_ transform {
     case e: Expression if e.semanticEquals(source) => destination
   })
+
+  private def selfReferenceOnly(e: Expression): Boolean = {
+    e.references.nonEmpty && e.references.subsetOf(outputSet) && e.deterministic
+  }
 }