{"action":"created","comment":{"url":"https://api.github.com/repos/apache/spark/pulls/comments/44199453","id":44199453,"diff_hunk":"@@ -213,3 +216,178 @@ object Utils {\n     case other => None\n   }\n }\n+\n+/**\n+ * This rule rewrites an aggregate query with multiple distinct clauses into an expanded double\n+ * aggregation in which the regular aggregation expressions and every distinct clause is aggregated\n+ * in a separate group. The results are then combined in a second aggregate.\n+ *\n+ * TODO Expression cannocalization\n+ * TODO Eliminate foldable expressions from distinct clauses.\n+ * TODO This eliminates all distinct expressions. We could safely pass one to the aggregate\n+ *      operator. Perhaps this is a good thing? It is much simpler to plan later on...\n+ */\n+object MultipleDistinctRewriter extends Rule[LogicalPlan] {\n+\n+  def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {\n+    case a: Aggregate => rewrite(a)\n+    case p => p\n+  }\n+\n+  def rewrite(a: Aggregate): Aggregate = {\n+\n+    // Collect all aggregate expressions.\n+    val aggExpressions = a.aggregateExpressions.flatMap { e =>\n+      e.collect {\n+        case ae: AggregateExpression2 => ae\n+      }\n+    }\n+\n+    // Extract distinct aggregate expressions.\n+    val distinctAggGroups = aggExpressions\n+      .filter(_.isDistinct)\n+      .groupBy(_.aggregateFunction.children.toSet)\n+\n+    // Only continue to rewrite if there is more than one distinct group.\n+    if (distinctAggGroups.size > 1) {\n+      // Create the attributes for the grouping id and the group by clause.\n+      val gid = new AttributeReference(\"gid\", IntegerType, false)()\n+      val groupByMap = a.groupingExpressions.collect {\n+        case ne: NamedExpression => ne -> ne.toAttribute\n+        case e => e -> new AttributeReference(e.prettyName, e.dataType, e.nullable)()\n+      }\n+      val groupByAttrs = groupByMap.map(_._2)\n+\n+      // Functions used to modify aggregate functions and their inputs.\n+      def evalWithinGroup(id: Literal, e: Expression) = If(EqualTo(gid, id), e, nullify(e))\n+      def patchAggregateFunctionChildren(\n+          af: AggregateFunction2,\n+          id: Literal,\n+          attrs: Map[Expression, Expression]): AggregateFunction2 = {\n+        af.withNewChildren(af.children.map { case afc =>\n+          evalWithinGroup(id, attrs(afc))\n+        }).asInstanceOf[AggregateFunction2]\n+      }\n+\n+      // Setup unique distinct aggregate children.\n+      val distinctAggChildren = distinctAggGroups.keySet.flatten.toSeq\n+      val distinctAggChildAttrMap = distinctAggChildren.map(expressionAttributePair).toMap\n+      val distinctAggChildAttrs = distinctAggChildAttrMap.values.toSeq\n+\n+      // Setup expand & aggregate operators for distinct aggregate expressions.\n+      val distinctAggOperatorMap = distinctAggGroups.toSeq.zipWithIndex.map {\n+        case ((group, expressions), i) =>\n+          val id = Literal(i + 1)\n+\n+          // Expand projection\n+          val projection = distinctAggChildren.map {\n+            case e if group.contains(e) => e\n+            case e => nullify(e)\n+          } :+ id\n+\n+          // Final aggregate\n+          val operators = expressions.map { e =>\n+            val af = e.aggregateFunction\n+            val naf = patchAggregateFunctionChildren(af, id, distinctAggChildAttrMap)\n+            (e, e.copy(aggregateFunction = naf, isDistinct = false))\n+          }\n+\n+          (projection, operators)\n+      }\n+\n+      // Setup expand for the 'regular' aggregate expressions.\n+      val regularAggExprs = aggExpressions.filter(!_.isDistinct)\n+      val regularAggChildren = regularAggExprs.flatMap(_.aggregateFunction.children).distinct\n+      val regularAggChildAttrMap = regularAggChildren.map(expressionAttributePair).toMap\n+\n+      // Setup aggregates for 'regular' aggregate expressions.\n+      val regularGroupId = Literal(0)\n+      val regularAggOperatorMap = regularAggExprs.map { e =>\n+        // Perform the actual aggregation in the initial aggregate.\n+        val af = patchAggregateFunctionChildren(\n+          e.aggregateFunction,\n+          regularGroupId,\n+          regularAggChildAttrMap)\n+        val a = Alias(e.copy(aggregateFunction = af), e.toString)()\n+\n+        // Get the result of the first aggregate in the last aggregate.\n+        val b = AggregateExpression2(\n+          aggregate.First(evalWithinGroup(regularGroupId, a.toAttribute), Literal(true)),\n+          mode = Complete,\n+          isDistinct = false)\n+\n+        // Some aggregate functions (COUNT) have the special property that they can return a\n+        // non-null result without any input. We need to make sure we return a result in this case.\n+        val c = af.defaultResult match {\n+          case Some(lit) => Coalesce(Seq(b, lit))\n+          case None => b\n+        }\n+\n+        (e, a, c)\n+      }\n+\n+      // Construct the regular aggregate input projection only if we need one.\n+      val regularAggProjection = if (regularAggExprs.nonEmpty) {\n+        Seq(a.groupingExpressions ++\n+          distinctAggChildren.map(nullify) ++\n+          Seq(regularGroupId) ++\n+          regularAggChildren)\n+      } else {\n+        Seq.empty[Seq[Expression]]\n+      }\n+\n+      // Construct the distinct aggregate input projections.\n+      val regularAggNulls = regularAggChildren.map(nullify)\n+      val distinctAggProjections = distinctAggOperatorMap.map {\n+        case (projection, _) =>\n+          a.groupingExpressions ++\n+            projection ++\n+            regularAggNulls\n+      }\n+\n+      // Construct the expand operator.\n+      val expand = Expand(\n+        regularAggProjection ++ distinctAggProjections,\n+        groupByAttrs ++ distinctAggChildAttrs ++ Seq(gid) ++ regularAggChildAttrMap.values.toSeq,\n+        a.child)\n+\n+      // Construct the first aggregate operator. This de-duplicates the all the children of\n+      // distinct operators, and applies the regular aggregate operators.\n+      val firstAggregateGroupBy = groupByAttrs ++ distinctAggChildAttrs :+ gid\n+      val firstAggregate = Aggregate(\n+        firstAggregateGroupBy,\n+        firstAggregateGroupBy ++ regularAggOperatorMap.map(_._2),\n+        expand)\n+\n+      // Construct the second aggregate\n+      val transformations: Map[Expression, Expression] =\n+        (distinctAggOperatorMap.flatMap(_._2) ++\n+          regularAggOperatorMap.map(e => (e._1, e._3))).toMap\n+\n+      val patchedAggExpressions = a.aggregateExpressions.map { e =>\n+        e.transformDown {\n+          case e: Expression =>\n+            // The same GROUP BY clauses can have different forms (different names for instance) in\n+            // the groupBy and aggregate expressions of an aggregate. This makes a map lookup\n+            // tricky. So we do a linear search for a semantically equal group by expression.","path":"sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Utils.scala","position":196,"original_position":196,"commit_id":"d3bdb2bb0096663af7d9faf4c0963a3df00065aa","original_commit_id":"d3bdb2bb0096663af7d9faf4c0963a3df00065aa","user":{"login":"marmbrus","id":527,"avatar_url":"https://avatars.githubusercontent.com/u/527?v=3","gravatar_id":"","url":"https://api.github.com/users/marmbrus","html_url":"https://github.com/marmbrus","followers_url":"https://api.github.com/users/marmbrus/followers","following_url":"https://api.github.com/users/marmbrus/following{/other_user}","gists_url":"https://api.github.com/users/marmbrus/gists{/gist_id}","starred_url":"https://api.github.com/users/marmbrus/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/marmbrus/subscriptions","organizations_url":"https://api.github.com/users/marmbrus/orgs","repos_url":"https://api.github.com/users/marmbrus/repos","events_url":"https://api.github.com/users/marmbrus/events{/privacy}","received_events_url":"https://api.github.com/users/marmbrus/received_events","type":"User","site_admin":false},"body":"We've talked about adding an `ExpressionMap` similar to `AttributeMap` in the past.  Seems like that would be useful here.","created_at":"2015-11-06T23:32:20Z","updated_at":"2015-11-06T23:32:20Z","html_url":"https://github.com/apache/spark/pull/9406#discussion_r44199453","pull_request_url":"https://api.github.com/repos/apache/spark/pulls/9406","_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/comments/44199453"},"html":{"href":"https://github.com/apache/spark/pull/9406#discussion_r44199453"},"pull_request":{"href":"https://api.github.com/repos/apache/spark/pulls/9406"}}},"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/9406","id":49411581,"html_url":"https://github.com/apache/spark/pull/9406","diff_url":"https://github.com/apache/spark/pull/9406.diff","patch_url":"https://github.com/apache/spark/pull/9406.patch","issue_url":"https://api.github.com/repos/apache/spark/issues/9406","number":9406,"state":"open","locked":false,"title":"[SPARK-9241][SQL] Supporting multiple DISTINCT columns (2) - Rewriting Rule","user":{"login":"hvanhovell","id":9616802,"avatar_url":"https://avatars.githubusercontent.com/u/9616802?v=3","gravatar_id":"","url":"https://api.github.com/users/hvanhovell","html_url":"https://github.com/hvanhovell","followers_url":"https://api.github.com/users/hvanhovell/followers","following_url":"https://api.github.com/users/hvanhovell/following{/other_user}","gists_url":"https://api.github.com/users/hvanhovell/gists{/gist_id}","starred_url":"https://api.github.com/users/hvanhovell/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/hvanhovell/subscriptions","organizations_url":"https://api.github.com/users/hvanhovell/orgs","repos_url":"https://api.github.com/users/hvanhovell/repos","events_url":"https://api.github.com/users/hvanhovell/events{/privacy}","received_events_url":"https://api.github.com/users/hvanhovell/received_events","type":"User","site_admin":false},"body":"The second PR for SPARK-9241, this adds support for multiple distinct columns to the new aggregation code path.\r\n\r\nThis PR solves the multiple DISTINCT column problem by rewriting these Aggregates into an Expand-Aggregate-Aggregate combination. See the [JIRA ticket](https://issues.apache.org/jira/browse/SPARK-9241) for some information on this. The advantages over the - competing - [first PR](https://github.com/apache/spark/pull/9280) are:\r\n- This can use the faster TungstenAggregate code path.\r\n- It is impossible to OOM due to an ```OpenHashSet``` allocating to much memory. However, this will multiply the number of input rows by the number of distinct clauses (plus one), and puts a lot more memory pressure on the aggregation code path itself.\r\n\r\nThe location of this Rule is a bit funny, and should probably change when the old aggregation path is changed.\r\n\r\ncc @yhuai - Could you also tell me where to add tests for this?","created_at":"2015-11-02T09:47:06Z","updated_at":"2015-11-06T23:32:20Z","closed_at":null,"merged_at":null,"merge_commit_sha":"6d68b639df79849aca8a2ac00478efbcf7db5db5","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/apache/spark/pulls/9406/commits","review_comments_url":"https://api.github.com/repos/apache/spark/pulls/9406/comments","review_comment_url":"https://api.github.com/repos/apache/spark/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/spark/issues/9406/comments","statuses_url":"https://api.github.com/repos/apache/spark/statuses/d3bdb2bb0096663af7d9faf4c0963a3df00065aa","head":{"label":"hvanhovell:SPARK-9241-rewriter","ref":"SPARK-9241-rewriter","sha":"d3bdb2bb0096663af7d9faf4c0963a3df00065aa","user":{"login":"hvanhovell","id":9616802,"avatar_url":"https://avatars.githubusercontent.com/u/9616802?v=3","gravatar_id":"","url":"https://api.github.com/users/hvanhovell","html_url":"https://github.com/hvanhovell","followers_url":"https://api.github.com/users/hvanhovell/followers","following_url":"https://api.github.com/users/hvanhovell/following{/other_user}","gists_url":"https://api.github.com/users/hvanhovell/gists{/gist_id}","starred_url":"https://api.github.com/users/hvanhovell/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/hvanhovell/subscriptions","organizations_url":"https://api.github.com/users/hvanhovell/orgs","repos_url":"https://api.github.com/users/hvanhovell/repos","events_url":"https://api.github.com/users/hvanhovell/events{/privacy}","received_events_url":"https://api.github.com/users/hvanhovell/received_events","type":"User","site_admin":false},"repo":{"id":35900688,"name":"spark","full_name":"hvanhovell/spark","owner":{"login":"hvanhovell","id":9616802,"avatar_url":"https://avatars.githubusercontent.com/u/9616802?v=3","gravatar_id":"","url":"https://api.github.com/users/hvanhovell","html_url":"https://github.com/hvanhovell","followers_url":"https://api.github.com/users/hvanhovell/followers","following_url":"https://api.github.com/users/hvanhovell/following{/other_user}","gists_url":"https://api.github.com/users/hvanhovell/gists{/gist_id}","starred_url":"https://api.github.com/users/hvanhovell/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/hvanhovell/subscriptions","organizations_url":"https://api.github.com/users/hvanhovell/orgs","repos_url":"https://api.github.com/users/hvanhovell/repos","events_url":"https://api.github.com/users/hvanhovell/events{/privacy}","received_events_url":"https://api.github.com/users/hvanhovell/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/hvanhovell/spark","description":"Mirror of Apache Spark","fork":true,"url":"https://api.github.com/repos/hvanhovell/spark","forks_url":"https://api.github.com/repos/hvanhovell/spark/forks","keys_url":"https://api.github.com/repos/hvanhovell/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/hvanhovell/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/hvanhovell/spark/teams","hooks_url":"https://api.github.com/repos/hvanhovell/spark/hooks","issue_events_url":"https://api.github.com/repos/hvanhovell/spark/issues/events{/number}","events_url":"https://api.github.com/repos/hvanhovell/spark/events","assignees_url":"https://api.github.com/repos/hvanhovell/spark/assignees{/user}","branches_url":"https://api.github.com/repos/hvanhovell/spark/branches{/branch}","tags_url":"https://api.github.com/repos/hvanhovell/spark/tags","blobs_url":"https://api.github.com/repos/hvanhovell/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/hvanhovell/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/hvanhovell/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/hvanhovell/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/hvanhovell/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/hvanhovell/spark/languages","stargazers_url":"https://api.github.com/repos/hvanhovell/spark/stargazers","contributors_url":"https://api.github.com/repos/hvanhovell/spark/contributors","subscribers_url":"https://api.github.com/repos/hvanhovell/spark/subscribers","subscription_url":"https://api.github.com/repos/hvanhovell/spark/subscription","commits_url":"https://api.github.com/repos/hvanhovell/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/hvanhovell/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/hvanhovell/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/hvanhovell/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/hvanhovell/spark/contents/{+path}","compare_url":"https://api.github.com/repos/hvanhovell/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/hvanhovell/spark/merges","archive_url":"https://api.github.com/repos/hvanhovell/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/hvanhovell/spark/downloads","issues_url":"https://api.github.com/repos/hvanhovell/spark/issues{/number}","pulls_url":"https://api.github.com/repos/hvanhovell/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/hvanhovell/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/hvanhovell/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/hvanhovell/spark/labels{/name}","releases_url":"https://api.github.com/repos/hvanhovell/spark/releases{/id}","created_at":"2015-05-19T18:29:31Z","updated_at":"2015-05-19T18:29:53Z","pushed_at":"2015-11-06T10:15:58Z","git_url":"git://github.com/hvanhovell/spark.git","ssh_url":"git@github.com:hvanhovell/spark.git","clone_url":"https://github.com/hvanhovell/spark.git","svn_url":"https://github.com/hvanhovell/spark","homepage":null,"size":127286,"stargazers_count":0,"watchers_count":0,"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":0,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"bc5d6c03893a9bd340d6b94d3550e25648412241","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}","created_at":"2014-02-25T08:00:08Z","updated_at":"2015-11-06T23:18:35Z","pushed_at":"2015-11-06T23:29:29Z","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":1653495,"stargazers_count":6057,"watchers_count":6057,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":5628,"mirror_url":"git://git.apache.org/spark.git","open_issues_count":367,"forks":5628,"open_issues":367,"watchers":6057,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/9406"},"html":{"href":"https://github.com/apache/spark/pull/9406"},"issue":{"href":"https://api.github.com/repos/apache/spark/issues/9406"},"comments":{"href":"https://api.github.com/repos/apache/spark/issues/9406/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/spark/pulls/9406/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/spark/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/spark/pulls/9406/commits"},"statuses":{"href":"https://api.github.com/repos/apache/spark/statuses/d3bdb2bb0096663af7d9faf4c0963a3df00065aa"}}}}