Github user yhuai commented on a diff in the pull request:
https://github.com/apache/spark/pull/5208#discussion_r28394599
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/Exchange.scala ---
@@ -120,27 +161,34 @@ case class Exchange(newPartitioning: Partitioning,
child: SparkPlan) extends Una
* Ensures that the
[[org.apache.spark.sql.catalyst.plans.physical.Partitioning Partitioning]]
* of input data meets the
* [[org.apache.spark.sql.catalyst.plans.physical.Distribution
Distribution]] requirements for
- * each operator by inserting [[Exchange]] Operators where required.
+ * each operator by inserting [[Exchange]] Operators where required. Also
ensure that the
+ * required input partition ordering requirements are met.
*/
-private[sql] case class AddExchange(sqlContext: SQLContext) extends
Rule[SparkPlan] {
+private[sql] case class EnsureRequirements(sqlContext: SQLContext) extends
Rule[SparkPlan] {
// TODO: Determine the number of partitions.
def numPartitions: Int = sqlContext.conf.numShufflePartitions
def apply(plan: SparkPlan): SparkPlan = plan.transformUp {
case operator: SparkPlan =>
- // Check if every child's outputPartitioning satisfies the
corresponding
+ // True iff every child's outputPartitioning satisfies the
corresponding
// required data distribution.
def meetsRequirements: Boolean =
- !operator.requiredChildDistribution.zip(operator.children).map {
+ operator.requiredChildDistribution.zip(operator.children).forall {
case (required, child) =>
val valid = child.outputPartitioning.satisfies(required)
logDebug(
s"${if (valid) "Valid" else "Invalid"} distribution," +
s"required: $required current:
${child.outputPartitioning}")
valid
- }.exists(!_)
+ }
- // Check if outputPartitionings of children are compatible with each
other.
+ // True iff any of the children are incorrectly sorted.
+ def needsAnySort: Boolean =
+ operator.requiredChildOrdering.zip(operator.children).exists {
+ case (required, child) => required.nonEmpty && required != child
--- End diff --
Seems you want `case (required, child) => required.nonEmpty && required !=
child. outputOrdering`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
