{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/15703","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/15703/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/15703/comments","events_url":"https://api.github.com/repos/apache/spark/issues/15703/events","html_url":"https://github.com/apache/spark/pull/15703","id":186419836,"number":15703,"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},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":22,"created_at":"2016-10-31T22:50:46Z","updated_at":"2016-11-11T00:04:53Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/15703","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"},"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. A byte array that backs 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"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/259844794","html_url":"https://github.com/apache/spark/pull/15703#issuecomment-259844794","issue_url":"https://api.github.com/repos/apache/spark/issues/15703","id":259844794,"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-11-11T00:04:52Z","updated_at":"2016-11-11T00:04:52Z","body":"Merged build finished. Test PASSed."}}
{"actor":{"display_login":"AmplabJenkins"}}