Skip to content

Commit f515cd2

Browse files
committed
yin's comment: outputOrdering, join suite refine
1 parent ec8061b commit f515cd2

File tree

2 files changed

+13
-3
lines changed

2 files changed

+13
-3
lines changed

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -185,7 +185,7 @@ private[sql] case class EnsureRequirements(sqlContext: SQLContext) extends Rule[
185185
// True iff any of the children are incorrectly sorted.
186186
def needsAnySort: Boolean =
187187
operator.requiredChildOrdering.zip(operator.children).exists {
188-
case (required, child) => required.nonEmpty && required != child
188+
case (required, child) => required.nonEmpty && required != child.outputOrdering
189189
}
190190

191191
// True iff outputPartitionings of children are compatible with each other.
@@ -233,7 +233,7 @@ private[sql] case class EnsureRequirements(sqlContext: SQLContext) extends Rule[
233233
}
234234
}
235235

236-
if (meetsRequirements && compatible && !needsAnySort) {
236+
if (meetsRequirements && compatible && !needsAnySort) {
237237
operator
238238
} else {
239239
// At least one child does not satisfies its required data distribution or

sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,6 @@ class JoinSuite extends QueryTest with BeforeAndAfterEach {
5151
case j: CartesianProduct => j
5252
case j: BroadcastNestedLoopJoin => j
5353
case j: BroadcastLeftSemiJoinHash => j
54-
case j: ShuffledHashJoin => j
5554
case j: SortMergeJoin => j
5655
}
5756

@@ -110,11 +109,22 @@ class JoinSuite extends QueryTest with BeforeAndAfterEach {
110109
cacheManager.clearCache()
111110
sql("CACHE TABLE testData")
112111

112+
val SORTMERGEJOIN_ENABLED: Boolean = conf.sortMergeJoinEnabled
113113
Seq(
114114
("SELECT * FROM testData join testData2 ON key = a", classOf[BroadcastHashJoin]),
115115
("SELECT * FROM testData join testData2 ON key = a and key = 2", classOf[BroadcastHashJoin]),
116116
("SELECT * FROM testData join testData2 ON key = a where key = 2", classOf[BroadcastHashJoin])
117117
).foreach { case (query, joinClass) => assertJoin(query, joinClass) }
118+
try {
119+
conf.setConf("spark.sql.planner.sortMergeJoin", "true")
120+
Seq(
121+
("SELECT * FROM testData join testData2 ON key = a", classOf[BroadcastHashJoin]),
122+
("SELECT * FROM testData join testData2 ON key = a and key = 2", classOf[BroadcastHashJoin]),
123+
("SELECT * FROM testData join testData2 ON key = a where key = 2", classOf[BroadcastHashJoin])
124+
).foreach { case (query, joinClass) => assertJoin(query, joinClass) }
125+
} finally {
126+
conf.setConf("spark.sql.planner.sortMergeJoin", SORTMERGEJOIN_ENABLED.toString)
127+
}
118128

119129
sql("UNCACHE TABLE testData")
120130
}

0 commit comments

Comments
 (0)