{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/13682","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/13682/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/13682/comments","events_url":"https://api.github.com/repos/apache/spark/issues/13682/events","html_url":"https://github.com/apache/spark/pull/13682","id":160351758,"number":13682,"title":"[SPARK-15888] [SQL] fix Python UDF with aggregate","user":{"login":"davies","id":40902,"avatar_url":"https://avatars.githubusercontent.com/u/40902?v=3","gravatar_id":"","url":"https://api.github.com/users/davies","html_url":"https://github.com/davies","followers_url":"https://api.github.com/users/davies/followers","following_url":"https://api.github.com/users/davies/following{/other_user}","gists_url":"https://api.github.com/users/davies/gists{/gist_id}","starred_url":"https://api.github.com/users/davies/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/davies/subscriptions","organizations_url":"https://api.github.com/users/davies/orgs","repos_url":"https://api.github.com/users/davies/repos","events_url":"https://api.github.com/users/davies/events{/privacy}","received_events_url":"https://api.github.com/users/davies/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":4,"created_at":"2016-06-15T07:01:25Z","updated_at":"2016-06-15T08:41:40Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/13682","html_url":"https://github.com/apache/spark/pull/13682","diff_url":"https://github.com/apache/spark/pull/13682.diff","patch_url":"https://github.com/apache/spark/pull/13682.patch"},"body":"## What changes were proposed in this pull request?\r\n\r\nAfter we move the ExtractPythonUDF rule into physical plan, Python UDF can't work on top of aggregate anymore, because they can't be evaluated before aggregate, should be evaluated after aggregate. This PR add another rule to extract these kind of Python UDF from logical aggregate, create a Project on top of Aggregate.\r\n\r\n## How was this patch tested?\r\n\r\nAdded regression tests. The plan of added test query looks like this:\r\n```\r\n== Parsed Logical Plan ==\r\n'Project [<lambda>('k, 's) AS t#26]\r\n+- Aggregate [<lambda>(key#5L)], [<lambda>(key#5L) AS k#17, sum(cast(<lambda>(value#6) as bigint)) AS s#22L]\r\n   +- LogicalRDD [key#5L, value#6]\r\n\r\n== Analyzed Logical Plan ==\r\nt: int\r\nProject [<lambda>(k#17, s#22L) AS t#26]\r\n+- Aggregate [<lambda>(key#5L)], [<lambda>(key#5L) AS k#17, sum(cast(<lambda>(value#6) as bigint)) AS s#22L]\r\n   +- LogicalRDD [key#5L, value#6]\r\n\r\n== Optimized Logical Plan ==\r\nProject [<lambda>(agg#29, agg#30L) AS t#26]\r\n+- Aggregate [<lambda>(key#5L)], [<lambda>(key#5L) AS agg#29, sum(cast(<lambda>(value#6) as bigint)) AS agg#30L]\r\n   +- LogicalRDD [key#5L, value#6]\r\n\r\n== Physical Plan ==\r\n*Project [pythonUDF0#37 AS t#26]\r\n+- BatchEvalPython [<lambda>(agg#29, agg#30L)], [agg#29, agg#30L, pythonUDF0#37]\r\n   +- *HashAggregate(key=[<lambda>(key#5L)#31], functions=[sum(cast(<lambda>(value#6) as bigint))], output=[agg#29,agg#30L])\r\n      +- Exchange hashpartitioning(<lambda>(key#5L)#31, 200)\r\n         +- *HashAggregate(key=[pythonUDF0#34 AS <lambda>(key#5L)#31], functions=[partial_sum(cast(pythonUDF1#35 as bigint))], output=[<lambda>(key#5L)#31,sum#33L])\r\n            +- BatchEvalPython [<lambda>(key#5L), <lambda>(value#6)], [key#5L, value#6, pythonUDF0#34, pythonUDF1#35]\r\n               +- Scan ExistingRDD[key#5L,value#6]\r\n```\r\n\r\n\r\n"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/226124767","html_url":"https://github.com/apache/spark/pull/13682#issuecomment-226124767","issue_url":"https://api.github.com/repos/apache/spark/issues/13682","id":226124767,"user":{"login":"AmplabJenkins","id":4031291,"avatar_url":"https://avatars.githubusercontent.com/u/4031291?v=3","gravatar_id":"","url":"https://api.github.com/users/AmplabJenkins","html_url":"https://github.com/AmplabJenkins","followers_url":"https://api.github.com/users/AmplabJenkins/followers","following_url":"https://api.github.com/users/AmplabJenkins/following{/other_user}","gists_url":"https://api.github.com/users/AmplabJenkins/gists{/gist_id}","starred_url":"https://api.github.com/users/AmplabJenkins/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/AmplabJenkins/subscriptions","organizations_url":"https://api.github.com/users/AmplabJenkins/orgs","repos_url":"https://api.github.com/users/AmplabJenkins/repos","events_url":"https://api.github.com/users/AmplabJenkins/events{/privacy}","received_events_url":"https://api.github.com/users/AmplabJenkins/received_events","type":"User","site_admin":false},"created_at":"2016-06-15T08:41:40Z","updated_at":"2016-06-15T08:41:40Z","body":"Merged build finished. Test PASSed."}}
{"actor":{"display_login":"AmplabJenkins"}}