Yin's comments

Author: sameeragarwal

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala

@@ -21,8 +21,7 @@ import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.trees.TreeNode
 import org.apache.spark.sql.types.{DataType, StructType}
 
-abstract class QueryPlan[PlanType <: TreeNode[PlanType]]
-  extends TreeNode[PlanType] with PredicateHelper {
+abstract class QueryPlan[PlanType <: TreeNode[PlanType]] extends TreeNode[PlanType] {
   self: PlanType =>
 
   def output: Seq[Attribute]
@@ -38,6 +37,11 @@ abstract class QueryPlan[PlanType <: TreeNode[PlanType]]
         constraint.references.nonEmpty && constraint.references.subsetOf(outputSet))
   }
 
+  /**
+   * Infers a set of `isNotNull` constraints from a given set of equality/comparison expressions.
+   * For e.g., if an expression is of the form (`a > 5`), this returns a constraint of the form
+   * `isNotNull(a)`
+   */
   private def constructIsNotNullConstraints(constraints: Set[Expression]): Set[Expression] = {
     // Currently we only propagate constraints if the condition consists of equality
     // and ranges. For all other cases, we return an empty set of constraints

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

@@ -301,7 +301,7 @@ abstract class LeafNode extends LogicalPlan {
 /**
  * A logical plan node with single child.
  */
-abstract class UnaryNode extends LogicalPlan with PredicateHelper {
+abstract class UnaryNode extends LogicalPlan {
   def child: LogicalPlan
 
   override def children: Seq[LogicalPlan] = child :: Nil

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

@@ -87,7 +87,8 @@ case class Generate(
   }
 }
 
-case class Filter(condition: Expression, child: LogicalPlan) extends UnaryNode {
+case class Filter(condition: Expression, child: LogicalPlan)
+  extends UnaryNode with PredicateHelper {
   override def output: Seq[Attribute] = child.output
 
   override protected def validConstraints: Set[Expression] = {
@@ -179,29 +180,35 @@ case class Union(children: Seq[LogicalPlan]) extends LogicalPlan {
     Statistics(sizeInBytes = sizeInBytes)
   }
 
-  def rewriteConstraints(
-      planA: LogicalPlan,
-      planB: LogicalPlan,
+  /**
+   * Maps the constraints containing a given (original) sequence of attributes to those with a
+   * given (reference) sequence of attributes. Given the nature of union, we expect that the
+   * mapping between the original and reference sequences are symmetric.
+   */
+  private def rewriteConstraints(
+      reference: Seq[Attribute],
+      original: Seq[Attribute],
       constraints: Set[Expression]): Set[Expression] = {
-    require(planA.output.size == planB.output.size)
-    val attributeRewrites = AttributeMap(planB.output.zip(planA.output))
+    require(reference.size == original.size)
+    val attributeRewrites = AttributeMap(original.zip(reference))
     constraints.map(_ transform {
       case a: Attribute => attributeRewrites(a)
     })
   }
 
   override protected def validConstraints: Set[Expression] = {
     children
-      .map(child => rewriteConstraints(children.head, child, child.constraints))
+      .map(child => rewriteConstraints(children.head.output, child.output, child.constraints))
       .reduce(_ intersect _)
   }
 }
 
 case class Join(
-  left: LogicalPlan,
-  right: LogicalPlan,
-  joinType: JoinType,
-  condition: Option[Expression]) extends BinaryNode {
+    left: LogicalPlan,
+    right: LogicalPlan,
+    joinType: JoinType,
+    condition: Option[Expression])
+  extends BinaryNode with PredicateHelper {
 
   override def output: Seq[Attribute] = {
     joinType match {
@@ -226,12 +233,11 @@ case class Join(
           .union(splitConjunctivePredicates(condition.get).toSet)
       case LeftSemi if condition.isDefined =>
         left.constraints
-          .union(right.constraints)
           .union(splitConjunctivePredicates(condition.get).toSet)
       case Inner =>
         left.constraints.union(right.constraints)
       case LeftSemi =>
-        left.constraints.union(right.constraints)
+        left.constraints
       case LeftOuter =>
         left.constraints
       case RightOuter =>