Skip to content
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

Fixes caching related JoinSuite failure #2960

Closed
wants to merge 1 commit into from
Closed

Fixes caching related JoinSuite failure #2960

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
117 changes: 62 additions & 55 deletions sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,17 +19,13 @@ package org.apache.spark.sql

import org.scalatest.BeforeAndAfterEach

import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
import org.apache.spark.sql.TestData._
import org.apache.spark.sql.catalyst.plans.JoinType
import org.apache.spark.sql.catalyst.plans.{LeftOuter, RightOuter, FullOuter, Inner, LeftSemi}
import org.apache.spark.sql.execution._
import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
import org.apache.spark.sql.catalyst.plans.{FullOuter, Inner, LeftOuter, RightOuter}
import org.apache.spark.sql.execution.joins._
import org.apache.spark.sql.test.TestSQLContext
import org.apache.spark.sql.test.TestSQLContext._

class JoinSuite extends QueryTest with BeforeAndAfterEach {

// Ensures tables are loaded.
TestData

Expand All @@ -41,54 +37,65 @@ class JoinSuite extends QueryTest with BeforeAndAfterEach {
assert(planned.size === 1)
}

test("join operator selection") {
def assertJoin(sqlString: String, c: Class[_]): Any = {
val rdd = sql(sqlString)
val physical = rdd.queryExecution.sparkPlan
val operators = physical.collect {
case j: ShuffledHashJoin => j
case j: HashOuterJoin => j
case j: LeftSemiJoinHash => j
case j: BroadcastHashJoin => j
case j: LeftSemiJoinBNL => j
case j: CartesianProduct => j
case j: BroadcastNestedLoopJoin => j
}

assert(operators.size === 1)
if (operators(0).getClass() != c) {
fail(s"$sqlString expected operator: $c, but got ${operators(0)}\n physical: \n$physical")
}
def assertJoin(sqlString: String, c: Class[_]): Any = {
val rdd = sql(sqlString)
val physical = rdd.queryExecution.sparkPlan
val operators = physical.collect {
case j: ShuffledHashJoin => j
case j: HashOuterJoin => j
case j: LeftSemiJoinHash => j
case j: BroadcastHashJoin => j
case j: LeftSemiJoinBNL => j
case j: CartesianProduct => j
case j: BroadcastNestedLoopJoin => j
}

assert(operators.size === 1)
if (operators(0).getClass() != c) {
fail(s"$sqlString expected operator: $c, but got ${operators(0)}\n physical: \n$physical")
}
}

val cases1 = Seq(
("SELECT * FROM testData left semi join testData2 ON key = a", classOf[LeftSemiJoinHash]),
("SELECT * FROM testData left semi join testData2", classOf[LeftSemiJoinBNL]),
("SELECT * FROM testData join testData2", classOf[CartesianProduct]),
("SELECT * FROM testData join testData2 where key=2", classOf[CartesianProduct]),
("SELECT * FROM testData left join testData2", classOf[CartesianProduct]),
("SELECT * FROM testData right join testData2", classOf[CartesianProduct]),
("SELECT * FROM testData full outer join testData2", classOf[CartesianProduct]),
("SELECT * FROM testData left join testData2 where key=2", classOf[CartesianProduct]),
("SELECT * FROM testData right join testData2 where key=2", classOf[CartesianProduct]),
("SELECT * FROM testData full outer join testData2 where key=2", classOf[CartesianProduct]),
("SELECT * FROM testData join testData2 where key>a", classOf[CartesianProduct]),
("SELECT * FROM testData full outer join testData2 where key>a", classOf[CartesianProduct]),
("SELECT * FROM testData join testData2 ON key = a", classOf[ShuffledHashJoin]),
("SELECT * FROM testData join testData2 ON key = a and key=2", classOf[ShuffledHashJoin]),
("SELECT * FROM testData join testData2 ON key = a where key=2", classOf[ShuffledHashJoin]),
("SELECT * FROM testData left join testData2 ON key = a", classOf[HashOuterJoin]),
("SELECT * FROM testData right join testData2 ON key = a where key=2",
test("join operator selection") {
clearCache()

Seq(
("SELECT * FROM testData LEFT SEMI JOIN testData2 ON key = a", classOf[LeftSemiJoinHash]),
("SELECT * FROM testData LEFT SEMI JOIN testData2", classOf[LeftSemiJoinBNL]),
("SELECT * FROM testData JOIN testData2", classOf[CartesianProduct]),
("SELECT * FROM testData JOIN testData2 WHERE key = 2", classOf[CartesianProduct]),
("SELECT * FROM testData LEFT JOIN testData2", classOf[CartesianProduct]),
("SELECT * FROM testData RIGHT JOIN testData2", classOf[CartesianProduct]),
("SELECT * FROM testData FULL OUTER JOIN testData2", classOf[CartesianProduct]),
("SELECT * FROM testData LEFT JOIN testData2 WHERE key = 2", classOf[CartesianProduct]),
("SELECT * FROM testData RIGHT JOIN testData2 WHERE key = 2", classOf[CartesianProduct]),
("SELECT * FROM testData FULL OUTER JOIN testData2 WHERE key = 2", classOf[CartesianProduct]),
("SELECT * FROM testData JOIN testData2 WHERE key > a", classOf[CartesianProduct]),
("SELECT * FROM testData FULL OUTER JOIN testData2 WHERE key > a", classOf[CartesianProduct]),
("SELECT * FROM testData JOIN testData2 ON key = a", classOf[ShuffledHashJoin]),
("SELECT * FROM testData JOIN testData2 ON key = a and key = 2", classOf[ShuffledHashJoin]),
("SELECT * FROM testData JOIN testData2 ON key = a where key = 2", classOf[ShuffledHashJoin]),
("SELECT * FROM testData LEFT JOIN testData2 ON key = a", classOf[HashOuterJoin]),
("SELECT * FROM testData RIGHT JOIN testData2 ON key = a where key = 2",
classOf[HashOuterJoin]),
("SELECT * FROM testData right join testData2 ON key = a and key=2",
("SELECT * FROM testData right join testData2 ON key = a and key = 2",
classOf[HashOuterJoin]),
("SELECT * FROM testData full outer join testData2 ON key = a", classOf[HashOuterJoin]),
("SELECT * FROM testData join testData2 ON key = a", classOf[ShuffledHashJoin]),
("SELECT * FROM testData join testData2 ON key = a and key=2", classOf[ShuffledHashJoin]),
("SELECT * FROM testData join testData2 ON key = a where key=2", classOf[ShuffledHashJoin])
// TODO add BroadcastNestedLoopJoin
)
cases1.foreach { c => assertJoin(c._1, c._2) }
("SELECT * FROM testData full outer join testData2 ON key = a", classOf[HashOuterJoin])
// TODO add BroadcastNestedLoopJoin
).foreach { case (query, joinClass) => assertJoin(query, joinClass) }
}

test("broadcasted hash join operator selection") {
clearCache()
sql("CACHE TABLE testData")

Seq(
("SELECT * FROM testData join testData2 ON key = a", classOf[BroadcastHashJoin]),
("SELECT * FROM testData join testData2 ON key = a and key = 2", classOf[BroadcastHashJoin]),
("SELECT * FROM testData join testData2 ON key = a where key = 2", classOf[BroadcastHashJoin])
).foreach { case (query, joinClass) => assertJoin(query, joinClass) }

sql("UNCACHE TABLE testData")
}

test("multiple-key equi-join is hash-join") {
Expand Down Expand Up @@ -171,7 +178,7 @@ class JoinSuite extends QueryTest with BeforeAndAfterEach {
(4, "D", 4, "d") ::
(5, "E", null, null) ::
(6, "F", null, null) :: Nil)

checkAnswer(
upperCaseData.join(lowerCaseData, LeftOuter, Some('n === 'N && 'n > 1)),
(1, "A", null, null) ::
Expand All @@ -180,7 +187,7 @@ class JoinSuite extends QueryTest with BeforeAndAfterEach {
(4, "D", 4, "d") ::
(5, "E", null, null) ::
(6, "F", null, null) :: Nil)

checkAnswer(
upperCaseData.join(lowerCaseData, LeftOuter, Some('n === 'N && 'N > 1)),
(1, "A", null, null) ::
Expand All @@ -189,7 +196,7 @@ class JoinSuite extends QueryTest with BeforeAndAfterEach {
(4, "D", 4, "d") ::
(5, "E", null, null) ::
(6, "F", null, null) :: Nil)

checkAnswer(
upperCaseData.join(lowerCaseData, LeftOuter, Some('n === 'N && 'l > 'L)),
(1, "A", 1, "a") ::
Expand Down Expand Up @@ -300,7 +307,7 @@ class JoinSuite extends QueryTest with BeforeAndAfterEach {
(4, "D", 4, "D") ::
(null, null, 5, "E") ::
(null, null, 6, "F") :: Nil)

checkAnswer(
left.join(right, FullOuter, Some(("left.N".attr === "right.N".attr) && ("left.N".attr !== 3))),
(1, "A", null, null) ::
Expand All @@ -310,7 +317,7 @@ class JoinSuite extends QueryTest with BeforeAndAfterEach {
(4, "D", 4, "D") ::
(null, null, 5, "E") ::
(null, null, 6, "F") :: Nil)

checkAnswer(
left.join(right, FullOuter, Some(("left.N".attr === "right.N".attr) && ("right.N".attr !== 3))),
(1, "A", null, null) ::
Expand Down
4 changes: 2 additions & 2 deletions sql/core/src/test/scala/org/apache/spark/sql/TestData.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,7 @@ object TestData {
UpperCaseData(3, "C") ::
UpperCaseData(4, "D") ::
UpperCaseData(5, "E") ::
UpperCaseData(6, "F") :: Nil)
UpperCaseData(6, "F") :: Nil).toSchemaRDD
upperCaseData.registerTempTable("upperCaseData")

case class LowerCaseData(n: Int, l: String)
Expand All @@ -89,7 +89,7 @@ object TestData {
LowerCaseData(1, "a") ::
LowerCaseData(2, "b") ::
LowerCaseData(3, "c") ::
LowerCaseData(4, "d") :: Nil)
LowerCaseData(4, "d") :: Nil).toSchemaRDD
Copy link
Contributor Author

Choose a reason for hiding this comment

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

These two changes are only introduced to calm down IntelliJ syntax highlighting...

lowerCaseData.registerTempTable("lowerCaseData")

case class ArrayData(data: Seq[Int], nestedData: Seq[Seq[Int]])
Expand Down