[SPARK-14664][SQL] Implement DecimalAggregates optimization for Window queries #12421

dongjoon-hyun · 2016-04-15T18:02:18Z

What changes were proposed in this pull request?

This PR aims to implement decimal aggregation optimization for window queries by improving existing DecimalAggregates. Historically, DecimalAggregates optimizer is designed to transform general sum/avg(decimal), but it breaks recently added windows queries like the followings. The following queries work well without the current DecimalAggregates optimizer.

Sum

scala> sql("select sum(a) over () from (select explode(array(1.0,2.0)) a) t").head
java.lang.RuntimeException: Unsupported window function: MakeDecimal((sum(UnscaledValue(a#31)),mode=Complete,isDistinct=false),12,1)
scala> sql("select sum(a) over () from (select explode(array(1.0,2.0)) a) t").explain()
== Physical Plan ==
WholeStageCodegen
:  +- Project [sum(a) OVER (  ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING)#23]
:     +- INPUT
+- Window [MakeDecimal((sum(UnscaledValue(a#21)),mode=Complete,isDistinct=false),12,1) windowspecdefinition(ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS sum(a) OVER (  ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING)#23]
   +- Exchange SinglePartition, None
      +- Generate explode([1.0,2.0]), false, false, [a#21]
         +- Scan OneRowRelation[]

Average

scala> sql("select avg(a) over () from (select explode(array(1.0,2.0)) a) t").head
java.lang.RuntimeException: Unsupported window function: cast(((avg(UnscaledValue(a#40)),mode=Complete,isDistinct=false) / 10.0) as decimal(6,5))
scala> sql("select avg(a) over () from (select explode(array(1.0,2.0)) a) t").explain()
== Physical Plan ==
WholeStageCodegen
:  +- Project [avg(a) OVER (  ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING)#44]
:     +- INPUT
+- Window [cast(((avg(UnscaledValue(a#42)),mode=Complete,isDistinct=false) / 10.0) as decimal(6,5)) windowspecdefinition(ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS avg(a) OVER (  ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING)#44]
   +- Exchange SinglePartition, None
      +- Generate explode([1.0,2.0]), false, false, [a#42]
         +- Scan OneRowRelation[]

After this PR, those queries work fine and new optimized physical plans look like the followings.

Sum

scala> sql("select sum(a) over () from (select explode(array(1.0,2.0)) a) t").explain()
== Physical Plan ==
WholeStageCodegen
:  +- Project [sum(a) OVER (  ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING)#35]
:     +- INPUT
+- Window [MakeDecimal((sum(UnscaledValue(a#33)),mode=Complete,isDistinct=false) windowspecdefinition(ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING),12,1) AS sum(a) OVER (  ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING)#35]
   +- Exchange SinglePartition, None
      +- Generate explode([1.0,2.0]), false, false, [a#33]
         +- Scan OneRowRelation[]

Average

scala> sql("select avg(a) over () from (select explode(array(1.0,2.0)) a) t").explain()
== Physical Plan ==
WholeStageCodegen
:  +- Project [avg(a) OVER (  ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING)#47]
:     +- INPUT
+- Window [cast(((avg(UnscaledValue(a#45)),mode=Complete,isDistinct=false) windowspecdefinition(ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) / 10.0) as decimal(6,5)) AS avg(a) OVER (  ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING)#47]
   +- Exchange SinglePartition, None
      +- Generate explode([1.0,2.0]), false, false, [a#45]
         +- Scan OneRowRelation[]

In this PR, SUM over window pattern matching is based on the code of @hvanhovell ; he should be credited for the work he did.

How was this patch tested?

Pass the Jenkins tests (with newly added testcases)

hvanhovell · 2016-04-15T18:17:50Z

@dongjoon-hyun good catch!

Can't we just make the optimizer respect the window expression? I.e. wrap the entire window expression in the Cast/MakeDecimal expression?

dongjoon-hyun · 2016-04-15T18:29:19Z

Oh, thank you for quick review. Actually, I tried to do first like that. There occurs exceptions about type mismatch due to the difference from input schema. So, it seems not so straightforward to me.

However, I'll try again according to your comments. Thank you, @hvanhovell !

SparkQA · 2016-04-15T19:35:43Z

Test build #55940 has finished for PR 12421 at commit 48003ee.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

dongjoon-hyun · 2016-04-21T23:01:00Z

Hi, @hvanhovell . According to your comments, I've worked this for a week, but I'm afraid of being a bottleneck.
I feel there is a gap between preventing and optimizing.
Do you think you could take over this issue? If you do that for me, I will close this PR after merging yours.
I think you are the best person to extend DecimalAggregate optimizer for Window query.

SparkQA · 2016-04-22T00:37:45Z

Test build #56602 has finished for PR 12421 at commit 7f64e62.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

dongjoon-hyun · 2016-04-22T17:01:23Z

Hi, @hvanhovell .
Could you give me some advice about how to handle this issue and this PR?

dongjoon-hyun · 2016-04-24T07:36:31Z

Rebased.

SparkQA · 2016-04-24T07:45:36Z

Test build #56840 has finished for PR 12421 at commit 39b5d14.

This patch fails MiMa tests.
This patch merges cleanly.
This patch adds no public classes.

hvanhovell · 2016-04-24T08:30:09Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

@@ -1321,7 +1321,9 @@ object DecimalAggregates extends Rule[LogicalPlan] {
  /** Maximum number of decimal digits representable precisely in a Double */
  private val MAX_DOUBLE_DIGITS = 15

-  def apply(plan: LogicalPlan): LogicalPlan = plan transformAllExpressions {
+  def apply(plan: LogicalPlan): LogicalPlan = plan transformExpressionsDown {
+    case we @ WindowExpression(AggregateExpression(_, _, _, _), _) => we


Does the following work for sum?

case we @ WindowExpression(ae @ AggregateExpression(Sum(e @ DecimalType.Expression(prec, scale)), _, _, _), _) if prec + 10 <= MAX_LONG_DIGITS => MakeDecimal(we.copy(windowFunction = ae.copy(aggregateFunction = Sum(UnscaledValue(e))), prec + 10, scale))

Oh, thank you for coming back, @hvanhovell . I tried that last week that, but it faced the following error in WindowExec.scala. (last week, it was Window.scala)

Unsupported window function: cast(((avg(UnscaledValue(a#14)),mode=Complete,isDistinct=false) / 10.0) as decimal(6,5)) java.lang.RuntimeException: Unsupported window function: cast(((avg(UnscaledValue(a#14)),mode=Complete,isDistinct=false) / 10.0) as decimal(6,5)) at scala.sys.package$.error(package.scala:27) at org.apache.spark.sql.execution.WindowExec$$anonfun$windowFrameExpressionFactoryPairs$1$$anonfun$apply$2.apply(WindowExec.scala:183)

After fixing that, there occured errors at compiling generated code (codeGen result) due to the mismatched type between input schema (Decimal to Long and vice versa). I did try to change the input schema or BoundedReferences, too.

Actually, this observation is the same one I mentioned 9 days ago. I tried other approaches in other ways. But nothing was a clean solution for this. So, I decided to ask your help.

dongjoon-hyun · 2016-04-25T05:23:48Z

Oh, my bad. Today, from your code, I tried again to implement DecimalAggregate optimizer for Windows and handled to implement it. I think I made some mistake during last week.

Thank you so much for your review and guiding, @hvanhovell ! After this PR passes the final Jenkins test, I will update the content of PR and JIRA, too.

SparkQA · 2016-04-25T06:21:39Z

Test build #56881 has finished for PR 12421 at commit d83d8f8.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

dongjoon-hyun · 2016-04-25T06:38:44Z

The failures are MemorySinkSuite. It seems to be irrelevant to this PR. I'll rebase to retrigger Jenkins.

[info] *** 2 TESTS FAILED ***
[error] Failed: Total 1962, Failed 2, Errors 0, Passed 1960, Ignored 28
[error] Failed tests:
[error]     org.apache.spark.sql.streaming.MemorySinkSuite
[error] (sql/test:test) sbt.TestsFailedException: Tests unsuccessful

SparkQA · 2016-04-25T08:07:39Z

Test build #56885 has finished for PR 12421 at commit c1d09d7.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

dongjoon-hyun · 2016-04-25T08:34:23Z

Hi, @hvanhovell .
Now, it's ready for review again.
Thank you so much. I couldn't fix this without your help.

SparkQA · 2016-04-26T00:13:18Z

Test build #56935 has finished for PR 12421 at commit 45beddd.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

dongjoon-hyun · 2016-04-26T00:22:24Z

More test cases are added.

dongjoon-hyun · 2016-04-26T04:58:40Z

Rebased.

SparkQA · 2016-04-26T06:25:54Z

Test build #56968 has finished for PR 12421 at commit 14fe6b9.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

dongjoon-hyun · 2016-04-26T17:02:09Z

Hi, @hvanhovell .
Could you review this PR when you have some time?

dongjoon-hyun · 2016-04-27T08:39:49Z

Hi, @rxin .
Could you review and merge this PR?
This fixes RuntimeException error for Window aggregation queries.

hvanhovell · 2016-04-27T09:13:34Z

@dongjoon-hyun I am gonna get to this today.

hvanhovell · 2016-04-27T11:07:33Z

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

+
+  test("SPARK-14664: Decimal sum/avg over window should work.") {
+    checkAnswer(
+      sqlContext.sql("select sum(a) over () from (select explode(array(1.0,2.0,3.0)) a) t"),


Note: you can also use select sum(a) over () from values 1.0, 2.0, 3.0 x(a)

Thank you. I will use this consice form.

hvanhovell · 2016-04-27T11:45:40Z

@dongjoon-hyun The PR is in pretty good shape. I left a few small comments/questions.

…w queries

dongjoon-hyun · 2016-04-27T18:00:21Z

@hvanhovell .
Thank you for squeezing your precious time for reviewing my PR several times.

SparkQA · 2016-04-27T19:17:17Z

Test build #57147 has finished for PR 12421 at commit 152c6c2.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2016-04-27T19:29:53Z

Test build #57149 has finished for PR 12421 at commit b2ac335.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

hvanhovell · 2016-04-27T19:35:27Z

LGTM

hvanhovell · 2016-04-27T19:35:40Z

Merging to master. Thanks!

dongjoon-hyun · 2016-04-27T19:38:41Z

Thank you so much, @hvanhovell !

hvanhovell reviewed Apr 24, 2016
View reviewed changes

dongjoon-hyun changed the title ~~[SPARK-14664][SQL] Fix DecimalAggregates optimizer not to break Window queries~~ [WIP][SPARK-14664][SQL] Fix DecimalAggregates optimizer not to break Window queries Apr 25, 2016

dongjoon-hyun changed the title ~~[WIP][SPARK-14664][SQL] Fix DecimalAggregates optimizer not to break Window queries~~ [SPARK-14664][SQL] Implement DecimalAggregates optimization for Window queries Apr 25, 2016

hvanhovell reviewed Apr 27, 2016
View reviewed changes

dongjoon-hyun added 3 commits April 27, 2016 10:50

[SPARK-14664][SQL] Fix DecimalAggregates optimizer not to break Windo…

5adc743

…w queries

Add more testcases

0373440

Address comments.

152c6c2

Simplify test sql statements.

b2ac335

asfgit closed this in af92299 Apr 27, 2016

dongjoon-hyun deleted the SPARK-14664 branch May 12, 2016 01:00

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-14664][SQL] Implement DecimalAggregates optimization for Window queries #12421

[SPARK-14664][SQL] Implement DecimalAggregates optimization for Window queries #12421

dongjoon-hyun commented Apr 15, 2016 •

edited

Loading

hvanhovell commented Apr 15, 2016

dongjoon-hyun commented Apr 15, 2016

SparkQA commented Apr 15, 2016

dongjoon-hyun commented Apr 21, 2016

SparkQA commented Apr 22, 2016

dongjoon-hyun commented Apr 22, 2016

dongjoon-hyun commented Apr 24, 2016

SparkQA commented Apr 24, 2016

hvanhovell Apr 24, 2016 •

edited

Loading

dongjoon-hyun Apr 24, 2016 •

edited

Loading

dongjoon-hyun commented Apr 25, 2016 •

edited

Loading

SparkQA commented Apr 25, 2016

dongjoon-hyun commented Apr 25, 2016

SparkQA commented Apr 25, 2016

dongjoon-hyun commented Apr 25, 2016

SparkQA commented Apr 26, 2016

dongjoon-hyun commented Apr 26, 2016

dongjoon-hyun commented Apr 26, 2016

SparkQA commented Apr 26, 2016

dongjoon-hyun commented Apr 26, 2016

dongjoon-hyun commented Apr 27, 2016

hvanhovell commented Apr 27, 2016

hvanhovell Apr 27, 2016

dongjoon-hyun Apr 27, 2016

hvanhovell commented Apr 27, 2016

dongjoon-hyun commented Apr 27, 2016

SparkQA commented Apr 27, 2016

SparkQA commented Apr 27, 2016

hvanhovell commented Apr 27, 2016

hvanhovell commented Apr 27, 2016

dongjoon-hyun commented Apr 27, 2016

[SPARK-14664][SQL] Implement DecimalAggregates optimization for Window queries #12421

[SPARK-14664][SQL] Implement DecimalAggregates optimization for Window queries #12421

Conversation

dongjoon-hyun commented Apr 15, 2016 • edited Loading

What changes were proposed in this pull request?

How was this patch tested?

hvanhovell commented Apr 15, 2016

dongjoon-hyun commented Apr 15, 2016

SparkQA commented Apr 15, 2016

dongjoon-hyun commented Apr 21, 2016

SparkQA commented Apr 22, 2016

dongjoon-hyun commented Apr 22, 2016

dongjoon-hyun commented Apr 24, 2016

SparkQA commented Apr 24, 2016

hvanhovell Apr 24, 2016 • edited Loading

Choose a reason for hiding this comment

dongjoon-hyun Apr 24, 2016 • edited Loading

Choose a reason for hiding this comment

dongjoon-hyun commented Apr 25, 2016 • edited Loading

SparkQA commented Apr 25, 2016

dongjoon-hyun commented Apr 25, 2016

SparkQA commented Apr 25, 2016

dongjoon-hyun commented Apr 25, 2016

SparkQA commented Apr 26, 2016

dongjoon-hyun commented Apr 26, 2016

dongjoon-hyun commented Apr 26, 2016

SparkQA commented Apr 26, 2016

dongjoon-hyun commented Apr 26, 2016

dongjoon-hyun commented Apr 27, 2016

hvanhovell commented Apr 27, 2016

hvanhovell Apr 27, 2016

Choose a reason for hiding this comment

dongjoon-hyun Apr 27, 2016

Choose a reason for hiding this comment

hvanhovell commented Apr 27, 2016

dongjoon-hyun commented Apr 27, 2016

SparkQA commented Apr 27, 2016

SparkQA commented Apr 27, 2016

hvanhovell commented Apr 27, 2016

hvanhovell commented Apr 27, 2016

dongjoon-hyun commented Apr 27, 2016

dongjoon-hyun commented Apr 15, 2016 •

edited

Loading

hvanhovell Apr 24, 2016 •

edited

Loading

dongjoon-hyun Apr 24, 2016 •

edited

Loading

dongjoon-hyun commented Apr 25, 2016 •

edited

Loading