[ https://issues.apache.org/jira/browse/SPARK-20636?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16722775#comment-16722775 ]
ASF GitHub Bot commented on SPARK-20636: ---------------------------------------- asfgit closed pull request #23222: [SPARK-20636] Add the rule TransposeWindow to the optimization batch URL: https://github.com/apache/spark/pull/23222 This is a PR merged from a forked repository. As GitHub hides the original diff on merge, it is displayed below for the sake of provenance: As this is a foreign pull request (from a fork), the diff is supplied below (as it won't show otherwise due to GitHub magic): diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala index f615757a837a1..3eb6bca6ec976 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala @@ -73,6 +73,7 @@ abstract class Optimizer(sessionCatalog: SessionCatalog) CombineLimits, CombineUnions, // Constant folding and strength reduction + TransposeWindow, NullPropagation, ConstantPropagation, FoldablePropagation, diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala index 9a5d5a9966ab7..9277dc6859247 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala @@ -20,6 +20,8 @@ package org.apache.spark.sql import org.scalatest.Matchers.the import org.apache.spark.TestUtils.{assertNotSpilled, assertSpilled} +import org.apache.spark.sql.catalyst.optimizer.TransposeWindow +import org.apache.spark.sql.execution.exchange.Exchange import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction, Window} import org.apache.spark.sql.functions._ import org.apache.spark.sql.internal.SQLConf @@ -668,18 +670,30 @@ class DataFrameWindowFunctionsSuite extends QueryTest with SharedSQLContext { ("S2", "P2", 300) ).toDF("sno", "pno", "qty") - val w1 = Window.partitionBy("sno") - val w2 = Window.partitionBy("sno", "pno") - - checkAnswer( - df.select($"sno", $"pno", $"qty", sum($"qty").over(w2).alias("sum_qty_2")) - .select($"sno", $"pno", $"qty", col("sum_qty_2"), sum("qty").over(w1).alias("sum_qty_1")), - Seq( - Row("S1", "P1", 100, 800, 800), - Row("S1", "P1", 700, 800, 800), - Row("S2", "P1", 200, 200, 500), - Row("S2", "P2", 300, 300, 500))) - + Seq(true, false).foreach { transposeWindowEnabled => + val excludedRules = if (transposeWindowEnabled) "" else TransposeWindow.ruleName + withSQLConf(SQLConf.OPTIMIZER_EXCLUDED_RULES.key -> excludedRules) { + val w1 = Window.partitionBy("sno") + val w2 = Window.partitionBy("sno", "pno") + + val select = df.select($"sno", $"pno", $"qty", sum($"qty").over(w2).alias("sum_qty_2")) + .select($"sno", $"pno", $"qty", col("sum_qty_2"), sum("qty").over(w1).alias("sum_qty_1")) + + val expectedNumExchanges = if (transposeWindowEnabled) 1 else 2 + val actualNumExchanges = select.queryExecution.executedPlan.collect { + case e: Exchange => e + }.length + assert(actualNumExchanges == expectedNumExchanges) + + checkAnswer( + select, + Seq( + Row("S1", "P1", 100, 800, 800), + Row("S1", "P1", 700, 800, 800), + Row("S2", "P1", 200, 200, 500), + Row("S2", "P2", 300, 300, 500))) + } + } } test("NaN and -0.0 in window partition keys") { ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Eliminate unnecessary shuffle with adjacent Window expressions > -------------------------------------------------------------- > > Key: SPARK-20636 > URL: https://issues.apache.org/jira/browse/SPARK-20636 > Project: Spark > Issue Type: Improvement > Components: Optimizer > Affects Versions: 2.1.1 > Reporter: Michael Styles > Assignee: Michael Styles > Priority: Major > Fix For: 3.0.0 > > > Consider the following example: > {noformat} > w1 = Window.partitionBy("sno") > w2 = Window.partitionBy("sno", "pno") > supply \ > .select('sno', 'pno', 'qty', F.sum('qty').over(w2).alias('sum_qty_2')) \ > .select('sno', 'pno', 'qty', F.col('sum_qty_2'), > F.sum('qty').over(w1).alias('sum_qty_1')) \ > .explain() > == Optimized Logical Plan == > Window [sum(qty#982L) windowspecdefinition(sno#980, ROWS BETWEEN UNBOUNDED > PRECEDING AND UNBOUNDED FOLLOWING) AS sum_qty_1#1112L], [sno#980] > +- Window [sum(qty#982L) windowspecdefinition(sno#980, pno#981, ROWS BETWEEN > UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS sum_qty_2#1105L], [sno#980, > pno#981] > +- Relation[sno#980,pno#981,qty#982L] parquet > == Physical Plan == > Window [sum(qty#982L) windowspecdefinition(sno#980, ROWS BETWEEN UNBOUNDED > PRECEDING AND UNBOUNDED FOLLOWING) AS sum_qty_1#1112L], [sno#980] > +- *Sort [sno#980 ASC NULLS FIRST], false, 0 > +- Exchange hashpartitioning(sno#980, 200) > +- Window [sum(qty#982L) windowspecdefinition(sno#980, pno#981, ROWS > BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS sum_qty_2#1105L], > [sno#980, pno#981] > +- *Sort [sno#980 ASC NULLS FIRST, pno#981 ASC NULLS FIRST], false, 0 > +- Exchange hashpartitioning(sno#980, pno#981, 200) > +- *FileScan parquet [sno#980,pno#981,qty#982L] ... > {noformat} > A more efficient query plan can be achieved by flipping the Window > expressions to eliminate an unnecessary shuffle as follows: > {noformat} > == Optimized Logical Plan == > Window [sum(qty#982L) windowspecdefinition(sno#980, pno#981, ROWS BETWEEN > UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS sum_qty_2#1087L], [sno#980, > pno#981] > +- Window [sum(qty#982L) windowspecdefinition(sno#980, ROWS BETWEEN UNBOUNDED > PRECEDING AND UNBOUNDED FOLLOWING) AS sum_qty_1#1085L], [sno#980] > +- Relation[sno#980,pno#981,qty#982L] parquet > == Physical Plan == > Window [sum(qty#982L) windowspecdefinition(sno#980, pno#981, ROWS BETWEEN > UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS sum_qty_2#1087L], [sno#980, > pno#981] > +- *Sort [sno#980 ASC NULLS FIRST, pno#981 ASC NULLS FIRST], false, 0 > +- Window [sum(qty#982L) windowspecdefinition(sno#980, ROWS BETWEEN > UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS sum_qty_1#1085L], [sno#980] > +- *Sort [sno#980 ASC NULLS FIRST], false, 0 > +- Exchange hashpartitioning(sno#980, 200) > +- *FileScan parquet [sno#980,pno#981,qty#982L] ... > {noformat} -- This message was sent by Atlassian JIRA (v7.6.3#76005) --------------------------------------------------------------------- To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org