{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/15317","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/15317/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/15317/comments","events_url":"https://api.github.com/repos/apache/spark/issues/15317/events","html_url":"https://github.com/apache/spark/pull/15317","id":180415529,"number":15317,"title":"[SPARK-17739][SQL] Collapse adjacent similar Window operators","user":{"login":"dongjoon-hyun","id":9700541,"avatar_url":"https://avatars.githubusercontent.com/u/9700541?v=3","gravatar_id":"","url":"https://api.github.com/users/dongjoon-hyun","html_url":"https://github.com/dongjoon-hyun","followers_url":"https://api.github.com/users/dongjoon-hyun/followers","following_url":"https://api.github.com/users/dongjoon-hyun/following{/other_user}","gists_url":"https://api.github.com/users/dongjoon-hyun/gists{/gist_id}","starred_url":"https://api.github.com/users/dongjoon-hyun/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/dongjoon-hyun/subscriptions","organizations_url":"https://api.github.com/users/dongjoon-hyun/orgs","repos_url":"https://api.github.com/users/dongjoon-hyun/repos","events_url":"https://api.github.com/users/dongjoon-hyun/events{/privacy}","received_events_url":"https://api.github.com/users/dongjoon-hyun/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":7,"created_at":"2016-09-30T22:30:29Z","updated_at":"2016-10-01T01:52:29Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/15317","html_url":"https://github.com/apache/spark/pull/15317","diff_url":"https://github.com/apache/spark/pull/15317.diff","patch_url":"https://github.com/apache/spark/pull/15317.patch"},"body":"## What changes were proposed in this pull request?\r\n\r\nCurrently, Spark does not collapse adjacent windows with the same partitioning and sorting. This PR implements `CollapseWindow` optimizer to do the followings.\r\n\r\n1. If the partition specs and order specs are the same, collapse into the parent.\r\n2. If the partition specs are the same and one order spec is a prefix of the other, collapse to the more specific one.\r\n\r\n\r\nFor example:\r\n```scala\r\nval df = spark.range(1000).select($\"id\" % 100 as \"grp\", $\"id\", rand() as \"col1\", rand() as \"col2\")\r\n\r\n// Add summary statistics for all columns\r\nimport org.apache.spark.sql.expressions.Window\r\nval cols = Seq(\"id\", \"col1\", \"col2\")\r\nval window = Window.partitionBy($\"grp\").orderBy($\"id\")\r\nval result = cols.foldLeft(df) { (base, name) =>\r\n  base.withColumn(s\"${name}_avg\", avg(col(name)).over(window))\r\n      .withColumn(s\"${name}_stddev\", stddev(col(name)).over(window))\r\n      .withColumn(s\"${name}_min\", min(col(name)).over(window))\r\n      .withColumn(s\"${name}_max\", max(col(name)).over(window))\r\n}\r\n```\r\n\r\n**Before**\r\n```scala\r\nscala> result.explain\r\n== Physical Plan ==\r\nWindow [max(col2#19) windowspecdefinition(grp#17L, id#14L ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS col2_max#234], [grp#17L], [id#14L ASC NULLS FIRST]\r\n+- Window [min(col2#19) windowspecdefinition(grp#17L, id#14L ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS col2_min#216], [grp#17L], [id#14L ASC NULLS FIRST]\r\n   +- Window [stddev_samp(col2#19) windowspecdefinition(grp#17L, id#14L ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS col2_stddev#191], [grp#17L], [id#14L ASC NULLS FIRST]\r\n      +- Window [avg(col2#19) windowspecdefinition(grp#17L, id#14L ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS col2_avg#167], [grp#17L], [id#14L ASC NULLS FIRST]\r\n         +- Window [max(col1#18) windowspecdefinition(grp#17L, id#14L ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS col1_max#152], [grp#17L], [id#14L ASC NULLS FIRST]\r\n            +- Window [min(col1#18) windowspecdefinition(grp#17L, id#14L ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS col1_min#138], [grp#17L], [id#14L ASC NULLS FIRST]\r\n               +- Window [stddev_samp(col1#18) windowspecdefinition(grp#17L, id#14L ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS col1_stddev#117], [grp#17L], [id#14L ASC NULLS FIRST]\r\n                  +- Window [avg(col1#18) windowspecdefinition(grp#17L, id#14L ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS col1_avg#97], [grp#17L], [id#14L ASC NULLS FIRST]\r\n                     +- Window [max(id#14L) windowspecdefinition(grp#17L, id#14L ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS id_max#86L], [grp#17L], [id#14L ASC NULLS FIRST]\r\n                        +- Window [min(id#14L) windowspecdefinition(grp#17L, id#14L ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS id_min#76L], [grp#17L], [id#14L ASC NULLS FIRST]\r\n                           +- *Project [grp#17L, id#14L, col1#18, col2#19, id_avg#26, id_stddev#42]\r\n                              +- Window [stddev_samp(_w0#59) windowspecdefinition(grp#17L, id#14L ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS id_stddev#42], [grp#17L], [id#14L ASC NULLS FIRST]\r\n                                 +- *Project [grp#17L, id#14L, col1#18, col2#19, id_avg#26, cast(id#14L as double) AS _w0#59]\r\n                                    +- Window [avg(id#14L) windowspecdefinition(grp#17L, id#14L ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS id_avg#26], [grp#17L], [id#14L ASC NULLS FIRST]\r\n                                       +- *Sort [grp#17L ASC NULLS FIRST, id#14L ASC NULLS FIRST], false, 0\r\n                                          +- Exchange hashpartitioning(grp#17L, 200)\r\n                                             +- *Project [(id#14L % 100) AS grp#17L, id#14L, rand(-6329949029880411066) AS col1#18, rand(-7251358484380073081) AS col2#19]\r\n                                                +- *Range (0, 1000, step=1, splits=Some(8))\r\n```\r\n\r\n**After**\r\n```scala\r\nscala> result.explain\r\n== Physical Plan ==\r\nWindow [max(col2#5) windowspecdefinition(grp#3L, id#0L ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS col2_max#220, min(col2#5) windowspecdefinition(grp#3L, id#0L ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS col2_min#202, stddev_samp(col2#5) windowspecdefinition(grp#3L, id#0L ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS col2_stddev#177, avg(col2#5) windowspecdefinition(grp#3L, id#0L ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS col2_avg#153, max(col1#4) windowspecdefinition(grp#3L, id#0L ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS col1_max#138, min(col1#4) windowspecdefinition(grp#3L, id#0L ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS col1_min#124, stddev_samp(col1#4) windowspecdefinition(grp#3L, id#0L ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS col1_stddev#103, avg(col1#4) windowspecdefinition(grp#3L, id#0L ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS col1_avg#83, max(id#0L) windowspecdefinition(grp#3L, id#0L ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS id_max#72L, min(id#0L) windowspecdefinition(grp#3L, id#0L ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS id_min#62L], [grp#3L], [id#0L ASC NULLS FIRST]\r\n+- *Project [grp#3L, id#0L, col1#4, col2#5, id_avg#12, id_stddev#28]\r\n   +- Window [stddev_samp(_w0#45) windowspecdefinition(grp#3L, id#0L ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS id_stddev#28], [grp#3L], [id#0L ASC NULLS FIRST]\r\n      +- *Project [grp#3L, id#0L, col1#4, col2#5, id_avg#12, cast(id#0L as double) AS _w0#45]\r\n         +- Window [avg(id#0L) windowspecdefinition(grp#3L, id#0L ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS id_avg#12], [grp#3L], [id#0L ASC NULLS FIRST]\r\n            +- *Sort [grp#3L ASC NULLS FIRST, id#0L ASC NULLS FIRST], false, 0\r\n               +- Exchange hashpartitioning(grp#3L, 200)\r\n                  +- *Project [(id#0L % 100) AS grp#3L, id#0L, rand(6537478539664068821) AS col1#4, rand(-8961093871295252795) AS col2#5]\r\n                     +- *Range (0, 1000, step=1, splits=Some(8))\r\n```\r\n\r\n## How was this patch tested?\r\n\r\nPass the Jenkins tests with a newly added testsuite.\r\n"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/250885656","html_url":"https://github.com/apache/spark/pull/15317#issuecomment-250885656","issue_url":"https://api.github.com/repos/apache/spark/issues/15317","id":250885656,"user":{"login":"SparkQA","id":8108735,"avatar_url":"https://avatars.githubusercontent.com/u/8108735?v=3","gravatar_id":"","url":"https://api.github.com/users/SparkQA","html_url":"https://github.com/SparkQA","followers_url":"https://api.github.com/users/SparkQA/followers","following_url":"https://api.github.com/users/SparkQA/following{/other_user}","gists_url":"https://api.github.com/users/SparkQA/gists{/gist_id}","starred_url":"https://api.github.com/users/SparkQA/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/SparkQA/subscriptions","organizations_url":"https://api.github.com/users/SparkQA/orgs","repos_url":"https://api.github.com/users/SparkQA/repos","events_url":"https://api.github.com/users/SparkQA/events{/privacy}","received_events_url":"https://api.github.com/users/SparkQA/received_events","type":"User","site_admin":false},"created_at":"2016-10-01T01:52:29Z","updated_at":"2016-10-01T01:52:29Z","body":"**[Test build #66198 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/66198/consoleFull)** for PR 15317 at commit [`4dbebda`](https://github.com/apache/spark/commit/4dbebda92ef065313d73492671f1be02cf5eed15)."}}
{"actor":{"display_login":"SparkQA"}}