{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/6413","labels_url":"https://api.github.com/repos/apache/spark/issues/6413/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/6413/comments","events_url":"https://api.github.com/repos/apache/spark/issues/6413/events","html_url":"https://github.com/apache/spark/pull/6413","id":81008373,"number":6413,"title":"[SPARK-7871][SQL][WIP] Improve the outputPartitioning for HashOuterJoin","user":{"login":"chenghao-intel","id":2367031,"avatar_url":"https://avatars.githubusercontent.com/u/2367031?v=3","gravatar_id":"","url":"https://api.github.com/users/chenghao-intel","html_url":"https://github.com/chenghao-intel","followers_url":"https://api.github.com/users/chenghao-intel/followers","following_url":"https://api.github.com/users/chenghao-intel/following{/other_user}","gists_url":"https://api.github.com/users/chenghao-intel/gists{/gist_id}","starred_url":"https://api.github.com/users/chenghao-intel/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/chenghao-intel/subscriptions","organizations_url":"https://api.github.com/users/chenghao-intel/orgs","repos_url":"https://api.github.com/users/chenghao-intel/repos","events_url":"https://api.github.com/users/chenghao-intel/events{/privacy}","received_events_url":"https://api.github.com/users/chenghao-intel/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":29,"created_at":"2015-05-26T15:16:54Z","updated_at":"2015-06-23T01:48:54Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/6413","html_url":"https://github.com/apache/spark/pull/6413","diff_url":"https://github.com/apache/spark/pull/6413.diff","patch_url":"https://github.com/apache/spark/pull/6413.patch"},"body":"https://issues.apache.org/jira/browse/SPARK-7871\r\nOptimize the Full outer join followed by another join with the same equi-join key.\r\nFor example:\r\n```\r\nexplain SELECT l.key, m.key, r.key from src l full outer join src m on l.key=m.key full outer join src r on l.key=r.key;\r\n\r\nBefore this PR: (4 Exchange, and include one Exchange for intermediate result repartitioning, this probably cause huge performance problem)\r\n== Physical Plan ==\r\nProject [key#1,key#3,key#5]\r\n HashOuterJoin [key#1], [key#5], FullOuter, None\r\n  Exchange (HashPartitioning [key#1], 200), []\r\n   Project [key#1,key#3]\r\n    HashOuterJoin [key#1], [key#3], FullOuter, None\r\n     Exchange (HashPartitioning [key#1], 200), []\r\n      HiveTableScan [key#1], (MetastoreRelation default, src, Some(l)), None\r\n     Exchange (HashPartitioning [key#3], 200), []\r\n      HiveTableScan [key#3], (MetastoreRelation default, src, Some(m)), None\r\n  Exchange (HashPartitioning [key#5], 200), []\r\n   HiveTableScan [key#5], (MetastoreRelation default, src, Some(r)), None\r\n\r\nApplied this PR (3 Exchange only for raw table data repartitioning)\r\n== Physical Plan ==\r\nProject [key#1,key#3,key#5]\r\n HashOuterJoin [key#1], [key#5], FullOuter, None\r\n  Project [key#1,key#3]\r\n   HashOuterJoin [key#1], [key#3], FullOuter, None\r\n    Exchange (HashPartitioning 200)\r\n     HiveTableScan [key#1], (MetastoreRelation default, src, Some(l)), None\r\n    Exchange (HashPartitioning 200)\r\n     HiveTableScan [key#3], (MetastoreRelation default, src, Some(m)), None\r\n  Exchange (HashPartitioning 200)\r\n   HiveTableScan [key#5], (MetastoreRelation default, src, Some(r)), None\r\n\r\nHowever, we don't want to change the logic for being followed by a GROUP BY, whose keys are exactly the same as the join key.\r\n\r\n\r\nexplain SELECT l.key, count(m.value) from src l full outer join src m on l.key=m.key group by l.key;\r\n== Physical Plan ==\r\nAggregate false, [key#9], [key#9,Coalesce(SUM(PartialCount#14L),0) AS _c1#7L]\r\n Exchange (HashPartitioning 200)\r\n  Aggregate true, [key#9], [key#9,COUNT(value#12) AS PartialCount#14L]\r\n   Project [key#9,value#12]\r\n    HashOuterJoin [key#9], [key#11], FullOuter, None\r\n     Exchange (HashPartitioning 200)\r\n      HiveTableScan [key#9], (MetastoreRelation default, src, Some(l)), None\r\n     Exchange (HashPartitioning 200)\r\n      HiveTableScan [key#11,value#12], (MetastoreRelation default, src, Some(m)), None\r\nEven the join key exactly the same with the group by key, however, the full outer join probably produce the null value for (l.key), hence we have to add another Exchange for Aggregate.\r\n```\r\n\r\nThis contains the code refactor for `exchange` in Spark SQL. It's a WIP PR and still need to do:\r\n- [ ] Enable more unit tests\r\n- [ ] Add More Scala Doc\r\n- [ ] Review all of the existed physical Plan for its `requiredDistribution` and `outputPartitioning`"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/114325452","html_url":"https://github.com/apache/spark/pull/6413#issuecomment-114325452","issue_url":"https://api.github.com/repos/apache/spark/issues/6413","id":114325452,"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":"2015-06-23T01:48:54Z","updated_at":"2015-06-23T01:48:54Z","body":"  [Test build #35498 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/35498/console) for   PR 6413 at commit [`b17a74d`](https://github.com/apache/spark/commit/b17a74dd019ee596e22b6cd0c6049d819cf4f82f).\n * This patch **fails Spark unit tests**.\n * This patch merges cleanly.\n * This patch adds the following public classes _(experimental)_:\n  * `case class ClusteredDistribution(`\n  * `sealed case class Partitioning(`\n"}}