You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by gatorsmile <gi...@git.apache.org> on 2016/02/22 05:13:51 UTC
[GitHub] spark pull request: [SPARK-13428] [SQL] Pushing Down Aggregate Exp...
GitHub user gatorsmile opened a pull request:
https://github.com/apache/spark/pull/11298
[SPARK-13428] [SQL] Pushing Down Aggregate Expressions in Sort into Aggregate
## What changes were proposed in this pull request?
When there exists the other operators between Sort and Aggregate, we are unable to push down the aggregate expressions in `Sort` into `Aggregate`. For example, in the following query,
```SQL
select area, sum(product) over () as c from windowData
where product > 3 group by area, product
having avg(month) > 0 order by area, avg(month), product
```
The parsed logical plan is like
```
'Sort ['area ASC,'avg('month) ASC,'product ASC], true
+- 'Filter cast(('avg('month) > 0) as boolean)
+- 'Aggregate ['area,'product], [unresolvedalias('area,None),unresolvedalias('sum('product) windowspecdefinition(UnspecifiedFrame) AS c#3,None)]
+- 'Filter ('product > 3)
+- 'UnresolvedRelation `windowData`, None
```
## How was the this patch tested?
Turn on a test case that `test("window function: Pushing aggregate Expressions in Sort to Aggregate")` exposes this issue.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/gatorsmile/spark aggExprInSortBy
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/11298.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #11298
----
commit 50071e2cf9c76fd1919c437500b82bfd287b5f95
Author: gatorsmile <ga...@gmail.com>
Date: 2016-02-22T03:57:45Z
pushing down aggregate expressions in Sort into Aggregate Operator
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13428] [SQL] Pushing Down Aggregate Exp...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11298#issuecomment-187031164
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13428] [SQL] Pushing Down Aggregate Exp...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/11298#discussion_r53585798
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -607,14 +607,21 @@ class Analyzer(
def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
// Skip sort with aggregate. This will be handled in ResolveAggregateFunctions
case sa @ Sort(_, _, child: Aggregate) => sa
--- End diff --
The above two issues have been covered in the rule `ResolveAggregateFunctions` when `Aggregate` is the child of `Sort`. Thus, the current PR keeps the existing logics without any change.
This PR is to introduce a fix for aggregate functions when there exists the other operators between `Aggregate` and `Sort`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13428] [SQL] Pushing Down Aggregate Exp...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11298#issuecomment-187031165
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/51658/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13428] [SQL] Pushing Down Aggregate Exp...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/11298#discussion_r53902571
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -607,14 +607,21 @@ class Analyzer(
def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
// Skip sort with aggregate. This will be handled in ResolveAggregateFunctions
case sa @ Sort(_, _, child: Aggregate) => sa
-
+ case sa @ Sort(_, _, _)
+ if sa.order.exists(so => ResolveAggregateFunctions.containsAggregate(so.child)) => sa
case s @ Sort(order, _, child) if !s.resolved && child.resolved =>
try {
val newOrder = order.map(resolveExpressionRecursively(_, child).asInstanceOf[SortOrder])
val requiredAttrs = AttributeSet(newOrder).filter(_.resolved)
val missingAttrs = requiredAttrs -- child.outputSet
- if (missingAttrs.nonEmpty) {
- // Add missing attributes and then project them away after the sort.
+ // resolve the unresolved functions for knowing if they are aggregate functions
+ val evaluatedSort =
+ ResolveFunctions.resolveFunctions(Sort(newOrder, s.global, child)).asInstanceOf[Sort]
+ if (evaluatedSort.order.exists(so =>
+ ResolveAggregateFunctions.containsAggregate(so.child))) {
+ // Skip sort with aggregate. This will be handled in ResolveAggregateFunctions
+ s.copy(order = evaluatedSort.order)
+ } else if (missingAttrs.nonEmpty) {
--- End diff --
nevermind. I got it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13428] [SQL] Pushing Down Aggregate Exp...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11298#issuecomment-187030953
**[Test build #51658 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/51658/consoleFull)** for PR 11298 at commit [`50071e2`](https://github.com/apache/spark/commit/50071e2cf9c76fd1919c437500b82bfd287b5f95).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13428] [SQL] Pushing Down Aggregate Exp...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11298#issuecomment-205729750
**[Test build #54975 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/54975/consoleFull)** for PR 11298 at commit [`50071e2`](https://github.com/apache/spark/commit/50071e2cf9c76fd1919c437500b82bfd287b5f95).
* This patch **fails R style tests**.
* This patch **does not merge cleanly**.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13428] [SQL] Pushing Down Aggregate Exp...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/11298#discussion_r53759608
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -607,14 +607,21 @@ class Analyzer(
def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
// Skip sort with aggregate. This will be handled in ResolveAggregateFunctions
case sa @ Sort(_, _, child: Aggregate) => sa
-
+ case sa @ Sort(_, _, _)
+ if sa.order.exists(so => ResolveAggregateFunctions.containsAggregate(so.child)) => sa
case s @ Sort(order, _, child) if !s.resolved && child.resolved =>
try {
val newOrder = order.map(resolveExpressionRecursively(_, child).asInstanceOf[SortOrder])
val requiredAttrs = AttributeSet(newOrder).filter(_.resolved)
val missingAttrs = requiredAttrs -- child.outputSet
- if (missingAttrs.nonEmpty) {
- // Add missing attributes and then project them away after the sort.
+ // resolve the unresolved functions for knowing if they are aggregate functions
+ val evaluatedSort =
+ ResolveFunctions.resolveFunctions(Sort(newOrder, s.global, child)).asInstanceOf[Sort]
+ if (evaluatedSort.order.exists(so =>
+ ResolveAggregateFunctions.containsAggregate(so.child))) {
--- End diff --
Can we move this part of resolving functions of Sort as another pattern matching case? So this original pattern matching case can be untouched? I think it will more clear.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13428] [SQL] Pushing Down Aggregate Exp...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/11298#discussion_r53761104
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -607,14 +607,21 @@ class Analyzer(
def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
// Skip sort with aggregate. This will be handled in ResolveAggregateFunctions
case sa @ Sort(_, _, child: Aggregate) => sa
-
+ case sa @ Sort(_, _, _)
+ if sa.order.exists(so => ResolveAggregateFunctions.containsAggregate(so.child)) => sa
case s @ Sort(order, _, child) if !s.resolved && child.resolved =>
try {
val newOrder = order.map(resolveExpressionRecursively(_, child).asInstanceOf[SortOrder])
val requiredAttrs = AttributeSet(newOrder).filter(_.resolved)
val missingAttrs = requiredAttrs -- child.outputSet
- if (missingAttrs.nonEmpty) {
- // Add missing attributes and then project them away after the sort.
+ // resolve the unresolved functions for knowing if they are aggregate functions
+ val evaluatedSort =
+ ResolveFunctions.resolveFunctions(Sort(newOrder, s.global, child)).asInstanceOf[Sort]
+ if (evaluatedSort.order.exists(so =>
+ ResolveAggregateFunctions.containsAggregate(so.child))) {
+ // Skip sort with aggregate. This will be handled in ResolveAggregateFunctions
+ s.copy(order = evaluatedSort.order)
+ } else if (missingAttrs.nonEmpty) {
--- End diff --
Will this branch be run? If there are missing attributes but because the sort order contains aggregate functions so it always hit the first condition?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13428] [SQL] Pushing Down Aggregate Exp...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on the pull request:
https://github.com/apache/spark/pull/11298#issuecomment-187005048
##### Issue 2: mixture of aliases and real columns in order by clause
```SQL
SELECT key as k, value as v, sum(value) FROM src GROUP BY key, value ORDER BY k, value
```
```
== Parsed Logical Plan ==
'Sort ['k ASC,'value ASC], true
+- 'Aggregate ['key,'value], [unresolvedalias('key AS k#47,None),unresolvedalias('value AS v#48,None),unresolvedalias('sum('value),None)]
+- 'UnresolvedRelation `src`, None
```
In the above case, we need to replace the actual column name `value` by the alias name `v`. Otherwise, we have to introduce `k` in all the nodes between `Sort` and `Aggregate`.
```
Project [k#47,v#48,_c2#54L]
+- Sort [k#47 ASC,v#48 ASC], true
+- Aggregate [key#45,value#46], [key#45 AS k#47,value#46 AS v#48,(sum(cast(value#46 as bigint)),mode=Complete,isDistinct=false) AS _c2#54L]
+- Subquery src
+- Project [_1#43 AS key#45,_2#44 AS value#46]
+- LocalRelation [_1#43,_2#44], [[1,1],[-1,1]]
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13428] [SQL] Pushing Down Aggregate Exp...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11298#issuecomment-205729777
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/54975/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13428] [SQL] Pushing Down Aggregate Exp...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11298#issuecomment-186998813
**[Test build #51658 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/51658/consoleFull)** for PR 11298 at commit [`50071e2`](https://github.com/apache/spark/commit/50071e2cf9c76fd1919c437500b82bfd287b5f95).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13428] [SQL] Pushing Down Aggregate Exp...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/11298#discussion_r53899820
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -607,14 +607,21 @@ class Analyzer(
def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
// Skip sort with aggregate. This will be handled in ResolveAggregateFunctions
case sa @ Sort(_, _, child: Aggregate) => sa
-
+ case sa @ Sort(_, _, _)
+ if sa.order.exists(so => ResolveAggregateFunctions.containsAggregate(so.child)) => sa
case s @ Sort(order, _, child) if !s.resolved && child.resolved =>
try {
val newOrder = order.map(resolveExpressionRecursively(_, child).asInstanceOf[SortOrder])
val requiredAttrs = AttributeSet(newOrder).filter(_.resolved)
val missingAttrs = requiredAttrs -- child.outputSet
- if (missingAttrs.nonEmpty) {
- // Add missing attributes and then project them away after the sort.
+ // resolve the unresolved functions for knowing if they are aggregate functions
+ val evaluatedSort =
+ ResolveFunctions.resolveFunctions(Sort(newOrder, s.global, child)).asInstanceOf[Sort]
+ if (evaluatedSort.order.exists(so =>
+ ResolveAggregateFunctions.containsAggregate(so.child))) {
--- End diff --
If we do not evaluate the functions, we are unable to know if this function is aggregate functions. Another idea might be a multi-pass solution.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13428] [SQL] Pushing Down Aggregate Exp...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/11298#discussion_r53902128
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -607,14 +607,21 @@ class Analyzer(
def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
// Skip sort with aggregate. This will be handled in ResolveAggregateFunctions
case sa @ Sort(_, _, child: Aggregate) => sa
-
+ case sa @ Sort(_, _, _)
+ if sa.order.exists(so => ResolveAggregateFunctions.containsAggregate(so.child)) => sa
case s @ Sort(order, _, child) if !s.resolved && child.resolved =>
try {
val newOrder = order.map(resolveExpressionRecursively(_, child).asInstanceOf[SortOrder])
val requiredAttrs = AttributeSet(newOrder).filter(_.resolved)
val missingAttrs = requiredAttrs -- child.outputSet
- if (missingAttrs.nonEmpty) {
- // Add missing attributes and then project them away after the sort.
+ // resolve the unresolved functions for knowing if they are aggregate functions
+ val evaluatedSort =
+ ResolveFunctions.resolveFunctions(Sort(newOrder, s.global, child)).asInstanceOf[Sort]
+ if (evaluatedSort.order.exists(so =>
+ ResolveAggregateFunctions.containsAggregate(so.child))) {
+ // Skip sort with aggregate. This will be handled in ResolveAggregateFunctions
+ s.copy(order = evaluatedSort.order)
+ } else if (missingAttrs.nonEmpty) {
--- End diff --
Looks like if the sort order has non empty missingAttrs, it will be ignored now.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13428] [SQL] Pushing Down Aggregate Exp...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11298#issuecomment-205728972
**[Test build #54975 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/54975/consoleFull)** for PR 11298 at commit [`50071e2`](https://github.com/apache/spark/commit/50071e2cf9c76fd1919c437500b82bfd287b5f95).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #11298: [SPARK-13428] [SQL] Pushing Down Aggregate Expres...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile closed the pull request at:
https://github.com/apache/spark/pull/11298
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13428] [SQL] Pushing Down Aggregate Exp...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on the pull request:
https://github.com/apache/spark/pull/11298#issuecomment-187009069
cc @davies @cloud-fan . This PR is like an incremental patch with few lines of code changes. I am not sure if the whole rewrite is worthy for resolving the above two issues only when there exist the other operators between `Sort` and `Aggregate`? After all, users can manually correct their query if they hit a resolution issue when missing attributes in Sort happen. We already cover most of cases. : )
Let me know if you want a whole rewrite or just an incremental fix. Thanks!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13428] [SQL] Pushing Down Aggregate Exp...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/11298#discussion_r53902037
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -607,14 +607,21 @@ class Analyzer(
def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
// Skip sort with aggregate. This will be handled in ResolveAggregateFunctions
case sa @ Sort(_, _, child: Aggregate) => sa
-
+ case sa @ Sort(_, _, _)
+ if sa.order.exists(so => ResolveAggregateFunctions.containsAggregate(so.child)) => sa
case s @ Sort(order, _, child) if !s.resolved && child.resolved =>
try {
val newOrder = order.map(resolveExpressionRecursively(_, child).asInstanceOf[SortOrder])
val requiredAttrs = AttributeSet(newOrder).filter(_.resolved)
val missingAttrs = requiredAttrs -- child.outputSet
- if (missingAttrs.nonEmpty) {
- // Add missing attributes and then project them away after the sort.
+ // resolve the unresolved functions for knowing if they are aggregate functions
+ val evaluatedSort =
+ ResolveFunctions.resolveFunctions(Sort(newOrder, s.global, child)).asInstanceOf[Sort]
+ if (evaluatedSort.order.exists(so =>
+ ResolveAggregateFunctions.containsAggregate(so.child))) {
+ // Skip sort with aggregate. This will be handled in ResolveAggregateFunctions
+ s.copy(order = evaluatedSort.order)
+ } else if (missingAttrs.nonEmpty) {
--- End diff --
But it is only possible when the condition `sort.order.exists(containsAggregate) || sort.child.isInstanceOf[Aggregate]` is true, right?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13428] [SQL] Pushing Down Aggregate Exp...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11298#issuecomment-205729764
Build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13428] [SQL] Pushing Down Aggregate Exp...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/11298#discussion_r53900047
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -607,14 +607,21 @@ class Analyzer(
def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
// Skip sort with aggregate. This will be handled in ResolveAggregateFunctions
case sa @ Sort(_, _, child: Aggregate) => sa
-
+ case sa @ Sort(_, _, _)
+ if sa.order.exists(so => ResolveAggregateFunctions.containsAggregate(so.child)) => sa
case s @ Sort(order, _, child) if !s.resolved && child.resolved =>
try {
val newOrder = order.map(resolveExpressionRecursively(_, child).asInstanceOf[SortOrder])
val requiredAttrs = AttributeSet(newOrder).filter(_.resolved)
val missingAttrs = requiredAttrs -- child.outputSet
- if (missingAttrs.nonEmpty) {
- // Add missing attributes and then project them away after the sort.
+ // resolve the unresolved functions for knowing if they are aggregate functions
+ val evaluatedSort =
+ ResolveFunctions.resolveFunctions(Sort(newOrder, s.global, child)).asInstanceOf[Sort]
+ if (evaluatedSort.order.exists(so =>
+ ResolveAggregateFunctions.containsAggregate(so.child))) {
+ // Skip sort with aggregate. This will be handled in ResolveAggregateFunctions
+ s.copy(order = evaluatedSort.order)
+ } else if (missingAttrs.nonEmpty) {
--- End diff --
In this case, the missing attributes will be handled by `ResolveAggregateFunctions`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13428] [SQL] Pushing Down Aggregate Exp...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on the pull request:
https://github.com/apache/spark/pull/11298#issuecomment-187001189
##### Issue 1:
```SQL
SELECT MAX(value) FROM src GROUP BY key + 1 ORDER BY key + 1
```
`key + 1` is not an aggregated function, but we still need to push it down into `Aggregate`, if we do not have `Project` and `Window` between `Sort` and `Aggregate`.
```
== Parsed Logical Plan ==
'Sort [('key + 1) ASC], true
+- 'Aggregate [('key + 1)], [unresolvedalias('MAX('value),None)]
+- 'UnresolvedRelation `src`, None
```
When that happens, we need to treat it as an aggregated function. The plan will be like:
```
Project [_c0#50]
+- Sort [aggOrder#51 ASC], true
+- Aggregate [(key#45 + 1)], [(max(value#46),mode=Complete,isDistinct=false) AS _c0#50,(key#45 + 1) AS aggOrder#51]
+- Subquery src
+- Project [_1#43 AS key#45,_2#44 AS value#46]
+- LocalRelation [_1#43,_2#44], [[1,1],[-1,1]]
```
Let me show another case that does not need to treat `key+1` as an aggregate function.
```SQL
SELECT value FROM src ORDER BY key + 1
```
```
== Parsed Logical Plan ==
'Sort [('key + 1) ASC], true
+- 'Project [unresolvedalias('value,None)]
+- 'UnresolvedRelation `src`, None
```
In this case, we just need to push `key` into `Project`.
```
Project [value#46]
+- Sort [(key#45 + 1) ASC], true
+- Project [value#46,key#45]
+- Subquery src
+- Project [_1#43 AS key#45,_2#44 AS value#46]
+- LocalRelation [_1#43,_2#44], [[1,1],[-1,1]]
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13428] [SQL] Pushing Down Aggregate Exp...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on the pull request:
https://github.com/apache/spark/pull/11298#issuecomment-186996652
This PR is an incremental fix based on the previous solution for resolving missing attributes in Sort. This is not a clean fix, I like. However, to do a clean fix, we need to rewrite a lot of code in `ResolveSortReferences`. I am not sure if we should do it. Will explain the issues I hit when resolving this one.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org