ODIn/spark-instrumented-optimizer
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

[SPARK-20186][SQL] BroadcastHint should use child's stats

Browse source 
## What changes were proposed in this pull request?

`BroadcastHint` should use child's statistics and set `isBroadcastable` to true.

## How was this patch tested?

Added a new stats estimation test for `BroadcastHint`.

Author: wangzhenhua <wangzhenhua@huawei.com>

Closes #17504 from wzhfy/broadcastHintEstimation.
This commit is contained in:
wangzhenhua 2017-04-01 22:19:08 +08:00 committed by Wenchen Fan
parent 89d6822f72
commit 2287f3d0b8
2 changed files with 21 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
View file

@ -383,7 +383,7 @@ case class BroadcastHint(child: LogicalPlan) extends UnaryNode {
// set isBroadcastable to true so the child will be broadcasted
override def computeStats(conf: CatalystConf): Statistics =
super.computeStats(conf).copy(isBroadcastable = true)
child.stats(conf).copy(isBroadcastable = true)
}
/**

21
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/statsEstimation/BasicStatsEstimationSuite.scala
View file

@ -35,6 +35,23 @@ class BasicStatsEstimationSuite extends StatsEstimationTestBase {
// row count * (overhead + column size)
size = Some(10 * (8 + 4)))
test("BroadcastHint estimation") {
val filter = Filter(Literal(true), plan)
val filterStatsCboOn = Statistics(sizeInBytes = 10 * (8 +4), isBroadcastable = false,
rowCount = Some(10), attributeStats = AttributeMap(Seq(attribute -> colStat)))
val filterStatsCboOff = Statistics(sizeInBytes = 10 * (8 +4), isBroadcastable = false)
checkStats(
filter,
expectedStatsCboOn = filterStatsCboOn,
expectedStatsCboOff = filterStatsCboOff)
val broadcastHint = BroadcastHint(filter)
checkStats(
broadcastHint,
expectedStatsCboOn = filterStatsCboOn.copy(isBroadcastable = true),
expectedStatsCboOff = filterStatsCboOff.copy(isBroadcastable = true))
}
test("limit estimation: limit < child's rowCount") {
val localLimit = LocalLimit(Literal(2), plan)
val globalLimit = GlobalLimit(Literal(2), plan)
@ -97,8 +114,10 @@ class BasicStatsEstimationSuite extends StatsEstimationTestBase {
plan: LogicalPlan,
expectedStatsCboOn: Statistics,
expectedStatsCboOff: Statistics): Unit = {
assert(plan.stats(conf.copy(cboEnabled = true)) == expectedStatsCboOn)
// Invalidate statistics
plan.invalidateStatsCache()
assert(plan.stats(conf.copy(cboEnabled = true)) == expectedStatsCboOn)
plan.invalidateStatsCache()
assert(plan.stats(conf.copy(cboEnabled = false)) == expectedStatsCboOff)
}