Skip to content

Commit e7d9a24

Browse files
wzhfymaropu
authored andcommitted
[SPARK-32817][SQL] DPP throws error when broadcast side is empty
### What changes were proposed in this pull request? In `SubqueryBroadcastExec.relationFuture`, if the `broadcastRelation` is an `EmptyHashedRelation`, then `broadcastRelation.keys()` will throw `UnsupportedOperationException`. ### Why are the changes needed? To fix a bug. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Added a new test. Closes #29671 from wzhfy/dpp_empty_broadcast. Authored-by: Zhenhua Wang <wzh_zju@163.com> Signed-off-by: Takeshi Yamamuro <yamamuro@apache.org>
1 parent bd3dc2f commit e7d9a24

3 files changed

Lines changed: 23 additions & 2 deletions

File tree

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1091,7 +1091,7 @@ case object EmptyHashedRelation extends HashedRelation {
10911091
override def keyIsUnique: Boolean = true
10921092

10931093
override def keys(): Iterator[InternalRow] = {
1094-
throw new UnsupportedOperationException
1094+
Iterator.empty
10951095
}
10961096

10971097
override def close(): Unit = {}

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

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1344,6 +1344,23 @@ abstract class DynamicPartitionPruningSuiteBase
13441344
}
13451345
}
13461346
}
1347+
1348+
test("SPARK-32817: DPP throws error when the broadcast side is empty") {
1349+
withSQLConf(
1350+
SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
1351+
SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "true") {
1352+
val df = sql(
1353+
"""
1354+
|SELECT * FROM fact_sk f
1355+
|JOIN dim_store s
1356+
|ON f.store_id = s.store_id WHERE s.country = 'XYZ'
1357+
""".stripMargin)
1358+
1359+
checkPartitionPruningPredicate(df, false, true)
1360+
1361+
checkAnswer(df, Nil)
1362+
}
1363+
}
13471364
}
13481365

13491366
class DynamicPartitionPruningSuiteAEOff extends DynamicPartitionPruningSuiteBase {

sql/core/src/test/scala/org/apache/spark/sql/execution/joins/HashedRelationSuite.scala

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -621,7 +621,7 @@ class HashedRelationSuite extends SharedSparkSession {
621621
}
622622
}
623623

624-
test("EmptyHashedRelation return null in get / getValue") {
624+
test("EmptyHashedRelation override methods behavior test") {
625625
val buildKey = Seq(BoundReference(0, LongType, false))
626626
val hashed = HashedRelation(Seq.empty[InternalRow].toIterator, buildKey, 1, mm)
627627
assert(hashed == EmptyHashedRelation)
@@ -631,6 +631,10 @@ class HashedRelationSuite extends SharedSparkSession {
631631
assert(hashed.get(key) == null)
632632
assert(hashed.getValue(0L) == null)
633633
assert(hashed.getValue(key) == null)
634+
635+
assert(hashed.keys().isEmpty)
636+
assert(hashed.keyIsUnique)
637+
assert(hashed.estimatedSize == 0)
634638
}
635639

636640
test("SPARK-32399: test methods related to key index") {

0 commit comments

Comments
 (0)