{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/7334","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/7334/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/7334/comments","events_url":"https://api.github.com/repos/apache/spark/issues/7334/events","html_url":"https://github.com/apache/spark/pull/7334","id":94166538,"number":7334,"title":"[SPARK-8964] [SQL] Use Exchange to perform shuffle in Limit","user":{"login":"JoshRosen","id":50748,"avatar_url":"https://avatars.githubusercontent.com/u/50748?v=3","gravatar_id":"","url":"https://api.github.com/users/JoshRosen","html_url":"https://github.com/JoshRosen","followers_url":"https://api.github.com/users/JoshRosen/followers","following_url":"https://api.github.com/users/JoshRosen/following{/other_user}","gists_url":"https://api.github.com/users/JoshRosen/gists{/gist_id}","starred_url":"https://api.github.com/users/JoshRosen/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/JoshRosen/subscriptions","organizations_url":"https://api.github.com/users/JoshRosen/orgs","repos_url":"https://api.github.com/users/JoshRosen/repos","events_url":"https://api.github.com/users/JoshRosen/events{/privacy}","received_events_url":"https://api.github.com/users/JoshRosen/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":29,"created_at":"2015-07-09T22:28:09Z","updated_at":"2016-02-05T21:07:28Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/7334","html_url":"https://github.com/apache/spark/pull/7334","diff_url":"https://github.com/apache/spark/pull/7334.diff","patch_url":"https://github.com/apache/spark/pull/7334.patch"},"body":"This patch changes the implementation of the physical `Limit` operator so that it relies on the `Exchange` operator to perform data movement rather than directly using `ShuffledRDD`. In addition to improving efficiency, this lays the necessary groundwork for further optimization of limit, such as limit pushdown or whole-stage codegen.\r\n\r\nAt a high-level, this replaces the old physical `Limit` operator with two new operators, `LocalLimit` and `GlobalLimit`. `LocalLimit` performs per-partition limits, while `GlobalLimit` applies the final limit to a single partition; `GlobalLimit`'s declares that its `requiredInputDistribution` is `SinglePartition`, which will cause the planner to use an `Exchange` to perform the appropriate shuffles. Thus, a logical `Limit` appearing in the middle of a query plan will be expanded into `LocalLimit -> Exchange to one partition -> GlobalLimit`.\r\n\r\nIn the old code, calling `someDataFrame.limit(100).collect()` or `someDataFrame.take(100)` would actually skip the shuffle and use a fast-path which used `executeTake()` in order to avoid computing all partitions in case only a small number of rows were requested. This patch preserves this optimization by treating logical `Limit` operators specially when they appear as the terminal operator in a query plan: if a `Limit` is the final operator, then we will plan a special `CollectLimit` physical operator which implements the old `take()`-based logic.\r\n\r\nIn order to be able to match on operators only at the root of the query plan, this patch introduces a special `ReturnAnswer` logical operator which functions similar to `BroadcastHint`: this dummy operator is inserted at the root of the optimized logical plan before invoking the physical planner, allowing the planner to pattern-match on it."},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/180554215","html_url":"https://github.com/apache/spark/pull/7334#issuecomment-180554215","issue_url":"https://api.github.com/repos/apache/spark/issues/7334","id":180554215,"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-02-05T21:07:27Z","updated_at":"2016-02-05T21:07:27Z","body":"**[Test build #50837 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/50837/consoleFull)** for PR 7334 at commit [`c4b0a53`](https://github.com/apache/spark/commit/c4b0a5303b4de328e11e8bfb79e8b3e795bd1174).\n * This patch **fails Spark unit tests**.\n * This patch merges cleanly.\n * This patch adds no public classes."}}