Skip to content

[SPARK-24495][SQL] EnsureRequirement returns wrong plan when reordering equal keys #21529

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
mgaido91 wants to merge 5 commits into from
Closed

[SPARK-24495][SQL] EnsureRequirement returns wrong plan when reordering equal keys #21529

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
06858cd
[SPARK-24495][SQL] EnsureRequirement returns worng plan when reorderi…
mgaido91 Jun 11, 2018
341f1b2
address comments
mgaido91 Jun 12, 2018
40abcff
address comments
mgaido91 Jun 13, 2018
6553c27
address comments
mgaido91 Jun 13, 2018
6ef4f0d
Merge branch 'master' of github.com:apache/spark into SPARK-24495
mgaido91 Jun 14, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@

package org.apache.spark.sql.execution.exchange

import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

import org.apache.spark.sql.catalyst.expressions._
Expand Down Expand Up @@ -227,9 +228,16 @@ case class EnsureRequirements(conf: SQLConf) extends Rule[SparkPlan] {
currentOrderOfKeys: Seq[Expression]): (Seq[Expression], Seq[Expression]) = {
val leftKeysBuffer = ArrayBuffer[Expression]()
val rightKeysBuffer = ArrayBuffer[Expression]()
val alreadyUsedIndexes = mutable.Set[Int]()
Copy link
Contributor

@cloud-fan cloud-fan Jun 11, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: maybe pickedIndexes?

val keysAndIndexes = currentOrderOfKeys.zipWithIndex

expectedOrderOfKeys.foreach(expression => {
val index = currentOrderOfKeys.indexWhere(e => e.semanticEquals(expression))
val index = keysAndIndexes.find { case (e, idx) =>
// As we may have the same key used many times, we need to filter out its occurrence we
// have already used.
e.semanticEquals(expression) && !alreadyUsedIndexes.contains(idx)
}.map(_._2).get
alreadyUsedIndexes += index
leftKeysBuffer.append(leftKeys(index))
rightKeysBuffer.append(rightKeys(index))
})
Expand Down
11 changes: 11 additions & 0 deletions sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -679,6 +679,17 @@ class PlannerSuite extends SharedSQLContext {
}
assert(rangeExecInZeroPartition.head.outputPartitioning == UnknownPartitioning(0))
}

test("SPARK-24495: EnsureRequirements can return wrong plan when reusing the same key in join") {
withSQLConf(("spark.sql.shuffle.partitions", "1"),
Copy link
Contributor

@cloud-fan cloud-fan Jun 12, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

let's not use hard coded config name, use SQLConf.SHUFFLE_PARTITIONS instead.

a nit: we usually use a -> b, c -> d, ... to specify config pairs.

("spark.sql.constraintPropagation.enabled", "false"),
("spark.sql.autoBroadcastJoinThreshold", "-1")) {
val df1 = spark.range(100)
Copy link
Contributor

@cloud-fan cloud-fan Jun 11, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we should make sure range has more than one partitions, otherwise we can't reproduce the bug.

val df2 = spark.range(100).select(($"id" * 2).as("b1"), (- $"id").as("b2"))
Copy link
Contributor

@cloud-fan cloud-fan Jun 12, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

($"id" * 2).as("b1") -> $"id".as("b1"), to minimize the test.

val res = df1.join(df2, $"id" === $"b1" && $"id" === $"b2")
assert(res.collect().sameElements(Array(Row(0, 0, 0))))
}
}
}

// Used for unit-testing EnsureRequirements
Expand Down