Update conditions for requiring child compatibility.

Author: JoshRosen

sql/core/src/main/scala/org/apache/spark/sql/execution/Exchange.scala

@@ -222,7 +222,10 @@ private[sql] case class EnsureRequirements(sqlContext: SQLContext) extends Rule[
    * output partitionings and add Exchanges to fix any detected incompatibilities.
    */
   private def ensureChildPartitioningsAreCompatible(operator: SparkPlan): SparkPlan = {
-    if (operator.requiresChildPartitioningsToBeCompatible) {
+    // If an operator has multiple children and the operator requires a specific child output
+    // distribution then we need to ensure that all children have compatible output partitionings.
+    if (operator.children.length > 1
+        && operator.requiredChildDistribution.toSet != Set(UnspecifiedDistribution)) {
       if (!Partitioning.allCompatible(operator.children.map(_.outputPartitioning))) {
         val newChildren = operator.children.zip(operator.requiredChildDistribution).map {
           case (child, requiredDistribution) =>

sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala

@@ -124,12 +124,6 @@ abstract class SparkPlan extends QueryPlan[SparkPlan] with Logging with Serializ
   /** Specifies sort order for each partition requirements on the input data for this operator. */
   def requiredChildOrdering: Seq[Seq[SortOrder]] = Seq.fill(children.size)(Nil)
 
-  /**
-   * Specifies whether this operator requires all of its children to have [[outputPartitioning]]s
-   * that are compatible with each other.
-   */
-  def requiresChildPartitioningsToBeCompatible: Boolean = false
-
   /** Specifies whether this operator outputs UnsafeRows */
   def outputsUnsafeRows: Boolean = false
 

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/LeftSemiJoinHash.scala

@@ -42,8 +42,6 @@ case class LeftSemiJoinHash(
   override def requiredChildDistribution: Seq[Distribution] =
     ClusteredDistribution(leftKeys) :: ClusteredDistribution(rightKeys) :: Nil
 
-  override def requiresChildPartitioningsToBeCompatible: Boolean = true
-
   protected override def doExecute(): RDD[InternalRow] = {
     right.execute().zipPartitions(left.execute()) { (buildIter, streamIter) =>
       if (condition.isEmpty) {

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledHashJoin.scala

@@ -46,8 +46,6 @@ case class ShuffledHashJoin(
   override def requiredChildDistribution: Seq[Distribution] =
     ClusteredDistribution(leftKeys) :: ClusteredDistribution(rightKeys) :: Nil
 
-  override def requiresChildPartitioningsToBeCompatible: Boolean = true
-
   protected override def doExecute(): RDD[InternalRow] = {
     buildPlan.execute().zipPartitions(streamedPlan.execute()) { (buildIter, streamIter) =>
       val hashed = HashedRelation(buildIter, buildSideKeyGenerator)

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledHashOuterJoin.scala

@@ -44,8 +44,6 @@ case class ShuffledHashOuterJoin(
   override def requiredChildDistribution: Seq[Distribution] =
     ClusteredDistribution(leftKeys) :: ClusteredDistribution(rightKeys) :: Nil
 
-  override def requiresChildPartitioningsToBeCompatible: Boolean = true
-
   override def outputPartitioning: Partitioning = joinType match {
     case LeftOuter => left.outputPartitioning
     case RightOuter => right.outputPartitioning

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoin.scala

@@ -48,8 +48,6 @@ case class SortMergeJoin(
   override def requiredChildDistribution: Seq[Distribution] =
     ClusteredDistribution(leftKeys) :: ClusteredDistribution(rightKeys) :: Nil
 
-  override def requiresChildPartitioningsToBeCompatible: Boolean = true
-
   override def outputOrdering: Seq[SortOrder] = requiredOrders(leftKeys)
 
   override def requiredChildOrdering: Seq[Seq[SortOrder]] =

sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala

@@ -214,7 +214,8 @@ class PlannerSuite extends SparkFunSuite with SQLTestUtils {
   // do they satisfy the distribution requirements? As a result, we need at least four test cases.
 
   private def assertDistributionRequirementsAreSatisfied(outputPlan: SparkPlan): Unit = {
-    if (outputPlan.requiresChildPartitioningsToBeCompatible) {
+    if (outputPlan.children.length > 1
+        && outputPlan.requiredChildDistribution.toSet != Set(UnspecifiedDistribution)) {
       val childPartitionings = outputPlan.children.map(_.outputPartitioning)
       if (!Partitioning.allCompatible(childPartitionings)) {
         fail(s"Partitionings are not compatible: $childPartitionings")
@@ -248,7 +249,6 @@ class PlannerSuite extends SparkFunSuite with SQLTestUtils {
         DummySparkPlan(outputPartitioning = leftPartitioning),
         DummySparkPlan(outputPartitioning = rightPartitioning)
       ),
-      requiresChildPartitioningsToBeCompatible = true,
       requiredChildDistribution = Seq(distribution, distribution),
       requiredChildOrdering = Seq(Seq.empty, Seq.empty)
     )
@@ -269,7 +269,6 @@ class PlannerSuite extends SparkFunSuite with SQLTestUtils {
         DummySparkPlan(outputPartitioning = HashPartitioning(clustering, 1)),
         DummySparkPlan(outputPartitioning = HashPartitioning(clustering, 2))
       ),
-      requiresChildPartitioningsToBeCompatible = true,
       requiredChildDistribution = Seq(distribution, distribution),
       requiredChildOrdering = Seq(Seq.empty, Seq.empty)
     )
@@ -288,7 +287,6 @@ class PlannerSuite extends SparkFunSuite with SQLTestUtils {
         DummySparkPlan(outputPartitioning = childPartitioning),
         DummySparkPlan(outputPartitioning = childPartitioning)
       ),
-      requiresChildPartitioningsToBeCompatible = true,
       requiredChildDistribution = Seq(distribution, distribution),
       requiredChildOrdering = Seq(Seq.empty, Seq.empty)
     )
@@ -309,7 +307,6 @@ class PlannerSuite extends SparkFunSuite with SQLTestUtils {
         DummySparkPlan(outputPartitioning = childPartitioning),
         DummySparkPlan(outputPartitioning = childPartitioning)
       ),
-      requiresChildPartitioningsToBeCompatible = true,
       requiredChildDistribution = Seq(distribution, distribution),
       requiredChildOrdering = Seq(Seq.empty, Seq.empty)
     )
@@ -333,7 +330,6 @@ class PlannerSuite extends SparkFunSuite with SQLTestUtils {
         DummySparkPlan(outputPartitioning = SinglePartition),
         DummySparkPlan(outputPartitioning = SinglePartition)
       ),
-      requiresChildPartitioningsToBeCompatible = true,
       requiredChildDistribution = Seq(distribution, distribution),
       requiredChildOrdering = Seq(outputOrdering, outputOrdering)
     )
@@ -352,7 +348,6 @@ private case class DummySparkPlan(
     override val children: Seq[SparkPlan] = Nil,
     override val outputOrdering: Seq[SortOrder] = Nil,
     override val outputPartitioning: Partitioning = UnknownPartitioning(0),
-    override val requiresChildPartitioningsToBeCompatible: Boolean = false,
     override val requiredChildDistribution: Seq[Distribution] = Nil,
     override val requiredChildOrdering: Seq[Seq[SortOrder]] = Nil
   ) extends SparkPlan {