You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2023/01/07 03:51:37 UTC
[GitHub] [spark] wankunde commented on a diff in pull request #39333: [SPARK-41805][SQL] Reuse expressions in WindowSpecDefinition
wankunde commented on code in PR #39333:
URL: https://github.com/apache/spark/pull/39333#discussion_r1063947944
##########
sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q36.sf100/simplified.txt:
##########
@@ -2,13 +2,13 @@ TakeOrderedAndProject [lochierarchy,i_category,rank_within_parent,gross_margin,i
WholeStageCodegen (7)
Project [gross_margin,i_category,i_class,lochierarchy,rank_within_parent]
InputAdapter
- Window [_w3,_w1,_w2]
+ Window [_w0,_w1,_w2]
WholeStageCodegen (6)
- Sort [_w1,_w2,_w3]
+ Sort [_w1,_w2,_w0]
InputAdapter
Exchange [_w1,_w2] #1
WholeStageCodegen (5)
- HashAggregate [i_category,i_class,spark_grouping_id,sum,sum] [sum(UnscaledValue(ss_net_profit)),sum(UnscaledValue(ss_ext_sales_price)),gross_margin,lochierarchy,_w1,_w2,_w3,sum,sum]
+ HashAggregate [i_category,i_class,spark_grouping_id,sum,sum] [sum(UnscaledValue(ss_net_profit)),sum(UnscaledValue(ss_ext_sales_price)),gross_margin,lochierarchy,_w0,_w1,_w2,sum,sum]
Review Comment:
* For window expr in q36:
```
rank()
OVER (
PARTITION BY grouping(i_category) + grouping(i_class),
CASE WHEN grouping(i_class) = 0
THEN i_category END
ORDER BY sum(ss_net_profit) / sum(ss_ext_sales_price) ASC) AS rank_within_parent
```
will be parsed into
```
{WindowExpression@13158} rank() windowspecdefinition((cast((shiftright(spark_grouping_id#117L, 1) & 1) as tinyint) AS grouping(i_category)#122 + cast((shiftright(spark_grouping_id#117L, 0) & 1) as tinyint) AS grouping(i_class)#123), CASE WHEN (cast(cast((shiftright(spark_grouping_id#117L, 0) & 1) as tinyint) AS grouping(i_class)#124 as int) = 0) THEN i_category#118 END, (sum(ss_net_profit#26) / sum(ss_ext_sales_price#19)) ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$()))
windowFunction = {Rank@9579} rank()
windowSpec = {WindowSpecDefinition@13160} windowspecdefinition((cast((shiftright(spark_grouping_id#117L, 1) & 1) as tinyint) AS grouping(i_category)#122 + cast((shiftright(spark_grouping_id#117L, 0) & 1) as tinyint) AS grouping(i_class)#123), CASE WHEN (cast(cast((shiftright(spark_grouping_id#117L, 0) & 1) as tinyint) AS grouping(i_class)#124 as int) = 0) THEN i_category#118 END, (sum(ss_net_profit#26) / sum(ss_ext_sales_price#19)) ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$()))
```
* Rule `ResolveWindowOrder` will add the **orderSpec** to rank function
```
object ResolveWindowOrder extends Rule[LogicalPlan] {
def apply(plan: LogicalPlan): LogicalPlan = plan.resolveExpressionsWithPruning(
_.containsPattern(WINDOW_EXPRESSION), ruleId) {
case WindowExpression(wf: WindowFunction, spec) if spec.orderSpec.isEmpty =>
throw QueryCompilationErrors.windowFunctionWithWindowFrameNotOrderedError(wf)
case WindowExpression(rank: RankLike, spec) if spec.resolved =>
val order = spec.orderSpec.map(_.child)
WindowExpression(rank.withOrder(order), spec)
}
}
```
so the window expression will be:
```
{WindowExpression@13219} rank((sum(ss_net_profit#26) / sum(ss_ext_sales_price#19))) windowspecdefinition((cast((shiftright(spark_grouping_id#117L, 1) & 1) as tinyint) AS grouping(i_category)#122 + cast((shiftright(spark_grouping_id#117L, 0) & 1) as tinyint) AS grouping(i_class)#123), CASE WHEN (cast(cast((shiftright(spark_grouping_id#117L, 0) & 1) as tinyint) AS grouping(i_class)#124 as int) = 0) THEN i_category#118 END, (sum(ss_net_profit#26) / sum(ss_ext_sales_price#19)) ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$()))
windowFunction = {Rank@13049} rank((sum(ss_net_profit#26) / sum(ss_ext_sales_price#19)))
windowSpec = {WindowSpecDefinition@13221} windowspecdefinition((cast((shiftright(spark_grouping_id#117L, 1) & 1) as tinyint) AS grouping(i_category)#122 + cast((shiftright(spark_grouping_id#117L, 0) & 1) as tinyint) AS grouping(i_class)#123), CASE WHEN (cast(cast((shiftright(spark_grouping_id#117L, 0) & 1) as tinyint) AS grouping(i_class)#124 as int) = 0) THEN i_category#118 END, (sum(ss_net_profit#26) / sum(ss_ext_sales_price#19)) ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$()))
```
* Before this PR, rule `ExtractWindowExpressions` will extract the child expr `(sum(ss_net_profit#26) / sum(ss_ext_sales_price#19))` in WindowFunction `Rank()` to `_w0` and extract the orderSpec expr `(sum(ss_net_profit#26) / sum(ss_ext_sales_price#19))` in WindowSpecDefinition to `_w3` again.
* After this PR, we will reuse the extract expr, and change `_w3` to `_w0` in q36 result.
I don't really understand why we need add **orderSpec** to **Rank** function. Could we remove
```
case WindowExpression(rank: RankLike, spec) if spec.resolved =>
val order = spec.orderSpec.map(_.child)
WindowExpression(rank.withOrder(order), spec)
```
from ResolveWindowOrder in the following PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org