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

[SPARK-27351][SQL] Wrong outputRows estimation after AggregateEstimation wit…

Browse source 
## What changes were proposed in this pull request?
The upper bound of group-by columns row number is to multiply distinct counts of group-by columns. However, column with only null value will cause the output row number to be 0 which is incorrect.
Ex:
col1 (distinct: 2, rowCount 2)
col2 (distinct: 0, rowCount 2)
=> group by col1, col2
Actual: output rows: 0
Expected: output rows: 2

## How was this patch tested?
According unit test has been added, plus manual test has been done in our tpcds benchmark environement.

Closes #24286 from pengbo/master.

Lead-authored-by: pengbo <bo.peng1019@gmail.com>
Co-authored-by: mingbo_pb <mingbo.pb@alibaba-inc.com>
Signed-off-by: Dongjoon Hyun <dhyun@apple.com>
(cherry picked from commit c58a4fed8d)
Signed-off-by: Dongjoon Hyun <dhyun@apple.com>
This commit is contained in:
pengbo 2019-04-15 15:37:07 -07:00 committed by Dongjoon Hyun
parent ede02b692f
commit 40668c53ed
2 changed files with 21 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/AggregateEstimation.scala
View file

@ -39,8 +39,16 @@ object AggregateEstimation {
// Multiply distinct counts of group-by columns. This is an upper bound, which assumes
// the data contains all combinations of distinct values of group-by columns.
var outputRows: BigInt = agg.groupingExpressions.foldLeft(BigInt(1))(
(res, expr) => res *
childStats.attributeStats(expr.asInstanceOf[Attribute]).distinctCount.get)
(res, expr) => {
val columnStat = childStats.attributeStats(expr.asInstanceOf[Attribute])
val distinctCount = columnStat.distinctCount.get
val distinctValue: BigInt = if (distinctCount == 0 && columnStat.nullCount.get > 0) {
1
} else {
distinctCount
}
res * distinctValue
})
outputRows = if (agg.groupingExpressions.isEmpty) {
// If there's no group-by columns, the output is a single row containing values of aggregate

12
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/statsEstimation/AggregateEstimationSuite.scala
View file

@ -38,7 +38,9 @@ class AggregateEstimationSuite extends StatsEstimationTestBase with PlanTest {
attr("key22") -> ColumnStat(distinctCount = Some(2), min = Some(10), max = Some(20),
nullCount = Some(0), avgLen = Some(4), maxLen = Some(4)),
attr("key31") -> ColumnStat(distinctCount = Some(0), min = None, max = None,
nullCount = Some(0), avgLen = Some(4), maxLen = Some(4))
nullCount = Some(0), avgLen = Some(4), maxLen = Some(4)),
attr("key32") -> ColumnStat(distinctCount = Some(0), min = None, max = None,
nullCount = Some(4), avgLen = Some(4), maxLen = Some(4))
))
private val nameToAttr: Map[String, Attribute] = columnInfo.map(kv => kv._1.name -> kv._1)
@ -92,6 +94,14 @@ class AggregateEstimationSuite extends StatsEstimationTestBase with PlanTest {
expectedOutputRowCount = 0)
}
test("group-by column with only null value") {
checkAggStats(
tableColumns = Seq("key22", "key32"),
tableRowCount = 6,
groupByColumns = Seq("key22", "key32"),
expectedOutputRowCount = nameToColInfo("key22")._2.distinctCount.get)
}
test("non-cbo estimation") {
val attributes = Seq("key12").map(nameToAttr)
val child = StatsTestPlan(