{"action":"created","comment":{"url":"https://api.github.com/repos/apache/spark/pulls/comments/88311998","pull_request_review_id":8889424,"id":88311998,"diff_hunk":"@@ -365,4 +380,66 @@ private[hive] case class HiveUDAFFunction(\n     val distinct = if (isDistinct) \"DISTINCT \" else \" \"\n     s\"$name($distinct${children.map(_.sql).mkString(\", \")})\"\n   }\n+\n+  override def createAggregationBuffer(): AggregationBuffer =\n+    partial1ModeEvaluator.getNewAggregationBuffer\n+\n+  @transient\n+  private lazy val inputProjection = new InterpretedProjection(children)\n+\n+  override def update(buffer: AggregationBuffer, input: InternalRow): Unit = {\n+    partial1ModeEvaluator.iterate(\n+      buffer, wrap(inputProjection(input), inputWrappers, cached, inputDataTypes))\n+  }\n+\n+  override def merge(buffer: AggregationBuffer, input: AggregationBuffer): Unit = {\n+    partial2ModeEvaluator.merge(buffer, partial1ModeEvaluator.terminatePartial(input))\n+  }\n+\n+  override def eval(buffer: AggregationBuffer): Any = {\n+    resultUnwrapper(finalModeEvaluator.terminate(buffer))\n+  }\n+\n+  override def serialize(buffer: AggregationBuffer): Array[Byte] = {\n+    aggBufferSerDe.serialize(buffer)\n+  }\n+\n+  override def deserialize(bytes: Array[Byte]): AggregationBuffer = {\n+    aggBufferSerDe.deserialize(bytes)\n+  }\n+\n+  // Helper class used to de/serialize Hive UDAF `AggregationBuffer` objects\n+  private class AggregationBufferSerDe {\n+    private val partialResultUnwrapper = unwrapperFor(partialResultInspector)\n+\n+    private val partialResultWrapper = wrapperFor(partialResultInspector, partialResultDataType)\n+\n+    private val projection = UnsafeProjection.create(Array(partialResultDataType))\n+\n+    private val mutableRow = new GenericInternalRow(1)\n+\n+    def serialize(buffer: AggregationBuffer): Array[Byte] = {","path":"sql/hive/src/main/scala/org/apache/spark/sql/hive/hiveUDFs.scala","position":261,"original_position":226,"commit_id":"ca3978c7f368225d349d91477ca13aa2d5b9a3fa","original_commit_id":"e88db5c9638678592f0f8836097e88d47558ffc1","user":{"login":"liancheng","id":230655,"avatar_url":"https://avatars.githubusercontent.com/u/230655?v=3","gravatar_id":"","url":"https://api.github.com/users/liancheng","html_url":"https://github.com/liancheng","followers_url":"https://api.github.com/users/liancheng/followers","following_url":"https://api.github.com/users/liancheng/following{/other_user}","gists_url":"https://api.github.com/users/liancheng/gists{/gist_id}","starred_url":"https://api.github.com/users/liancheng/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/liancheng/subscriptions","organizations_url":"https://api.github.com/users/liancheng/orgs","repos_url":"https://api.github.com/users/liancheng/repos","events_url":"https://api.github.com/users/liancheng/events{/privacy}","received_events_url":"https://api.github.com/users/liancheng/received_events","type":"User","site_admin":false},"body":"Done.","created_at":"2016-11-16T19:26:26Z","updated_at":"2016-11-16T19:26:26Z","html_url":"https://github.com/apache/spark/pull/15703#discussion_r88311998","pull_request_url":"https://api.github.com/repos/apache/spark/pulls/15703","_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/comments/88311998"},"html":{"href":"https://github.com/apache/spark/pull/15703#discussion_r88311998"},"pull_request":{"href":"https://api.github.com/repos/apache/spark/pulls/15703"}}},"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/15703","id":91705685,"html_url":"https://github.com/apache/spark/pull/15703","diff_url":"https://github.com/apache/spark/pull/15703.diff","patch_url":"https://github.com/apache/spark/pull/15703.patch","issue_url":"https://api.github.com/repos/apache/spark/issues/15703","number":15703,"state":"open","locked":false,"title":"[SPARK-18186] Migrate HiveUDAFFunction to TypedImperativeAggregate for partial aggregation support","user":{"login":"liancheng","id":230655,"avatar_url":"https://avatars.githubusercontent.com/u/230655?v=3","gravatar_id":"","url":"https://api.github.com/users/liancheng","html_url":"https://github.com/liancheng","followers_url":"https://api.github.com/users/liancheng/followers","following_url":"https://api.github.com/users/liancheng/following{/other_user}","gists_url":"https://api.github.com/users/liancheng/gists{/gist_id}","starred_url":"https://api.github.com/users/liancheng/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/liancheng/subscriptions","organizations_url":"https://api.github.com/users/liancheng/orgs","repos_url":"https://api.github.com/users/liancheng/repos","events_url":"https://api.github.com/users/liancheng/events{/privacy}","received_events_url":"https://api.github.com/users/liancheng/received_events","type":"User","site_admin":false},"body":"## What changes were proposed in this pull request?\r\n\r\nWhile being evaluated in Spark SQL, Hive UDAFs don't support partial aggregation. This PR migrates `HiveUDAFFunction`s to `TypedImperativeAggregate`, which already provides partial aggregation support for aggregate functions that may use arbitrary Java objects as aggregation states.\r\n\r\nThe following snippet shows the effect of this PR:\r\n\r\n```scala\r\nimport org.apache.hadoop.hive.ql.udf.generic.GenericUDAFMax\r\nsql(s\"CREATE FUNCTION hive_max AS '${classOf[GenericUDAFMax].getName}'\")\r\n\r\nspark.range(100).createOrReplaceTempView(\"t\")\r\n\r\n// A query using both Spark SQL native `max` and Hive `max`\r\nsql(s\"SELECT max(id), hive_max(id) FROM t\").explain()\r\n```\r\n\r\nBefore this PR:\r\n\r\n```\r\n== Physical Plan ==\r\nSortAggregate(key=[], functions=[max(id#1L), default.hive_max(default.hive_max, HiveFunctionWrapper(org.apache.hadoop.hive.ql.udf.generic.GenericUDAFMax,org.apache.hadoop.hive.ql.udf.generic.GenericUDAFMax@7475f57e), id#1L, false, 0, 0)])\r\n+- Exchange SinglePartition\r\n   +- *Range (0, 100, step=1, splits=Some(1))\r\n```\r\n\r\nAfter this PR:\r\n\r\n```\r\n== Physical Plan ==\r\nSortAggregate(key=[], functions=[max(id#1L), default.hive_max(default.hive_max, HiveFunctionWrapper(org.apache.hadoop.hive.ql.udf.generic.GenericUDAFMax,org.apache.hadoop.hive.ql.udf.generic.GenericUDAFMax@5e18a6a7), id#1L, false, 0, 0)])\r\n+- Exchange SinglePartition\r\n   +- SortAggregate(key=[], functions=[partial_max(id#1L), partial_default.hive_max(default.hive_max, HiveFunctionWrapper(org.apache.hadoop.hive.ql.udf.generic.GenericUDAFMax,org.apache.hadoop.hive.ql.udf.generic.GenericUDAFMax@5e18a6a7), id#1L, false, 0, 0)])\r\n      +- *Range (0, 100, step=1, splits=Some(1))\r\n```\r\n\r\nThe tricky part of the PR is mostly about updating and passing around aggregation states of `HiveUDAFFunction`s since the aggregation state of a Hive UDAF may appear in three different forms. Let's take a look at the testing `MockUDAF` added in this PR as an example. This UDAF computes the count of non-null values together with the count of nulls of a given column. Its aggregation state may appear as the following forms at different time:\r\n\r\n1. A `MockUDAFBuffer`, which is a concrete subclass of `GenericUDAFEvaluator.AggregationBuffer`\r\n\r\n   The form used by Hive UDAF API. This form is required by the following scenarios:\r\n\r\n   - Calling `GenericUDAFEvaluator.iterate()` to update an existing aggregation state with new input values.\r\n   - Calling `GenericUDAFEvaluator.terminate()` to get the final aggregated value from an existing aggregation state.\r\n   - Calling `GenericUDAFEvaluator.merge()` to merge other aggregation states into an existing aggregation state.\r\n\r\n     The existing aggregation state to be updated must be in this form.\r\n\r\n   Conversions:\r\n\r\n   - To form 2:\r\n\r\n     `GenericUDAFEvaluator.terminatePartial()`\r\n\r\n   - To form 3:\r\n\r\n     Convert to form 2 first, and then to 3.\r\n\r\n2. An `Object[]` array containing two `java.lang.Long` values.\r\n\r\n   The form used to interact with Hive's `ObjectInspector`s. This form is required by the following scenarios:\r\n\r\n   - Calling `GenericUDAFEvaluator.terminatePartial()` to convert an existing aggregation state in form 1 to form 2.\r\n   - Calling `GenericUDAFEvaluator.merge()` to merge other aggregation states into an existing aggregation state.\r\n\r\n     The input aggregation state must be in this form.\r\n\r\n   Conversions:\r\n\r\n   - To form 1:\r\n\r\n     No direct method. Have to create an empty `AggregationBuffer` and merge it into the empty buffer.\r\n\r\n   - To form 3:\r\n\r\n     `unwrapperFor()`/`unwrap()` method of `HiveInspectors`\r\n\r\n3. The byte array that holds data of an `UnsafeRow` with two `LongType` fields.\r\n\r\n   The form used by Spark SQL to shuffle partial aggregation results. This form is required because `TypedImperativeAggregate` always asks its subclasses to serialize their aggregation states into a byte array.\r\n\r\n   Conversions:\r\n\r\n   - To form 1:\r\n\r\n     Convert to form 2 first, and then to 1.\r\n\r\n   - To form 2:\r\n\r\n     `wrapperFor()`/`wrap()` method of `HiveInspectors`\r\n\r\n## How was this patch tested?\r\n\r\nNew test suite `HiveUDAFSuite` is added.\r\n","created_at":"2016-10-31T22:50:46Z","updated_at":"2016-11-16T19:26:26Z","closed_at":null,"merged_at":null,"merge_commit_sha":"e7fe5df12cd3b92c993268eba1a1d0b71e6a0ef6","assignee":null,"assignees":[],"milestone":null,"commits_url":"https://api.github.com/repos/apache/spark/pulls/15703/commits","review_comments_url":"https://api.github.com/repos/apache/spark/pulls/15703/comments","review_comment_url":"https://api.github.com/repos/apache/spark/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/spark/issues/15703/comments","statuses_url":"https://api.github.com/repos/apache/spark/statuses/ca3978c7f368225d349d91477ca13aa2d5b9a3fa","head":{"label":"liancheng:partial-agg-hive-udaf","ref":"partial-agg-hive-udaf","sha":"ca3978c7f368225d349d91477ca13aa2d5b9a3fa","user":{"login":"liancheng","id":230655,"avatar_url":"https://avatars.githubusercontent.com/u/230655?v=3","gravatar_id":"","url":"https://api.github.com/users/liancheng","html_url":"https://github.com/liancheng","followers_url":"https://api.github.com/users/liancheng/followers","following_url":"https://api.github.com/users/liancheng/following{/other_user}","gists_url":"https://api.github.com/users/liancheng/gists{/gist_id}","starred_url":"https://api.github.com/users/liancheng/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/liancheng/subscriptions","organizations_url":"https://api.github.com/users/liancheng/orgs","repos_url":"https://api.github.com/users/liancheng/repos","events_url":"https://api.github.com/users/liancheng/events{/privacy}","received_events_url":"https://api.github.com/users/liancheng/received_events","type":"User","site_admin":false},"repo":{"id":17232535,"name":"spark","full_name":"liancheng/spark","owner":{"login":"liancheng","id":230655,"avatar_url":"https://avatars.githubusercontent.com/u/230655?v=3","gravatar_id":"","url":"https://api.github.com/users/liancheng","html_url":"https://github.com/liancheng","followers_url":"https://api.github.com/users/liancheng/followers","following_url":"https://api.github.com/users/liancheng/following{/other_user}","gists_url":"https://api.github.com/users/liancheng/gists{/gist_id}","starred_url":"https://api.github.com/users/liancheng/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/liancheng/subscriptions","organizations_url":"https://api.github.com/users/liancheng/orgs","repos_url":"https://api.github.com/users/liancheng/repos","events_url":"https://api.github.com/users/liancheng/events{/privacy}","received_events_url":"https://api.github.com/users/liancheng/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/liancheng/spark","description":"Mirror of Apache Spark","fork":true,"url":"https://api.github.com/repos/liancheng/spark","forks_url":"https://api.github.com/repos/liancheng/spark/forks","keys_url":"https://api.github.com/repos/liancheng/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/liancheng/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/liancheng/spark/teams","hooks_url":"https://api.github.com/repos/liancheng/spark/hooks","issue_events_url":"https://api.github.com/repos/liancheng/spark/issues/events{/number}","events_url":"https://api.github.com/repos/liancheng/spark/events","assignees_url":"https://api.github.com/repos/liancheng/spark/assignees{/user}","branches_url":"https://api.github.com/repos/liancheng/spark/branches{/branch}","tags_url":"https://api.github.com/repos/liancheng/spark/tags","blobs_url":"https://api.github.com/repos/liancheng/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/liancheng/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/liancheng/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/liancheng/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/liancheng/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/liancheng/spark/languages","stargazers_url":"https://api.github.com/repos/liancheng/spark/stargazers","contributors_url":"https://api.github.com/repos/liancheng/spark/contributors","subscribers_url":"https://api.github.com/repos/liancheng/spark/subscribers","subscription_url":"https://api.github.com/repos/liancheng/spark/subscription","commits_url":"https://api.github.com/repos/liancheng/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/liancheng/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/liancheng/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/liancheng/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/liancheng/spark/contents/{+path}","compare_url":"https://api.github.com/repos/liancheng/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/liancheng/spark/merges","archive_url":"https://api.github.com/repos/liancheng/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/liancheng/spark/downloads","issues_url":"https://api.github.com/repos/liancheng/spark/issues{/number}","pulls_url":"https://api.github.com/repos/liancheng/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/liancheng/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/liancheng/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/liancheng/spark/labels{/name}","releases_url":"https://api.github.com/repos/liancheng/spark/releases{/id}","deployments_url":"https://api.github.com/repos/liancheng/spark/deployments","created_at":"2014-02-27T01:39:03Z","updated_at":"2015-03-31T08:56:24Z","pushed_at":"2016-11-16T19:17:33Z","git_url":"git://github.com/liancheng/spark.git","ssh_url":"git@github.com:liancheng/spark.git","clone_url":"https://github.com/liancheng/spark.git","svn_url":"https://github.com/liancheng/spark","homepage":null,"size":241346,"stargazers_count":1,"watchers_count":1,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":1,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"e0deee1f7df31177cfc14bbb296f0baa372f473d","user":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":17165658,"name":"spark","full_name":"apache/spark","owner":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/apache/spark","description":"Mirror of Apache Spark","fork":false,"url":"https://api.github.com/repos/apache/spark","forks_url":"https://api.github.com/repos/apache/spark/forks","keys_url":"https://api.github.com/repos/apache/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/spark/teams","hooks_url":"https://api.github.com/repos/apache/spark/hooks","issue_events_url":"https://api.github.com/repos/apache/spark/issues/events{/number}","events_url":"https://api.github.com/repos/apache/spark/events","assignees_url":"https://api.github.com/repos/apache/spark/assignees{/user}","branches_url":"https://api.github.com/repos/apache/spark/branches{/branch}","tags_url":"https://api.github.com/repos/apache/spark/tags","blobs_url":"https://api.github.com/repos/apache/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/spark/languages","stargazers_url":"https://api.github.com/repos/apache/spark/stargazers","contributors_url":"https://api.github.com/repos/apache/spark/contributors","subscribers_url":"https://api.github.com/repos/apache/spark/subscribers","subscription_url":"https://api.github.com/repos/apache/spark/subscription","commits_url":"https://api.github.com/repos/apache/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/spark/contents/{+path}","compare_url":"https://api.github.com/repos/apache/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/spark/merges","archive_url":"https://api.github.com/repos/apache/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/spark/downloads","issues_url":"https://api.github.com/repos/apache/spark/issues{/number}","pulls_url":"https://api.github.com/repos/apache/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/spark/labels{/name}","releases_url":"https://api.github.com/repos/apache/spark/releases{/id}","deployments_url":"https://api.github.com/repos/apache/spark/deployments","created_at":"2014-02-25T08:00:08Z","updated_at":"2016-11-16T18:16:29Z","pushed_at":"2016-11-16T19:19:31Z","git_url":"git://github.com/apache/spark.git","ssh_url":"git@github.com:apache/spark.git","clone_url":"https://github.com/apache/spark.git","svn_url":"https://github.com/apache/spark","homepage":null,"size":610622,"stargazers_count":10747,"watchers_count":10747,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":10034,"mirror_url":"git://git.apache.org/spark.git","open_issues_count":444,"forks":10034,"open_issues":444,"watchers":10747,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/15703"},"html":{"href":"https://github.com/apache/spark/pull/15703"},"issue":{"href":"https://api.github.com/repos/apache/spark/issues/15703"},"comments":{"href":"https://api.github.com/repos/apache/spark/issues/15703/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/spark/pulls/15703/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/spark/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/spark/pulls/15703/commits"},"statuses":{"href":"https://api.github.com/repos/apache/spark/statuses/ca3978c7f368225d349d91477ca13aa2d5b9a3fa"}}}}
{"actor":{"display_login":"liancheng"}}