You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "bersprockets (via GitHub)" <gi...@apache.org> on 2024/10/23 19:22:11 UTC
[PR] [SPARK-50091][SQL] Handle case of aggregates in left-hand operand of IN-subquery [spark]
bersprockets opened a new pull request, #48627:
URL: https://github.com/apache/spark/pull/48627
### What changes were proposed in this pull request?
This PR adds code to `RewritePredicateSubquery#apply` to explicitly handle the case where an `Aggregate` node contains an aggregate expression in the left-hand operand of an IN-subquery expression. The explicit handler moves the IN-subquery expressions out of the `Aggregate` and into a parent `Project` node. The `Aggregate` will continue to perform the aggregations that were used as an operand to the IN-subquery expression, but will not include the IN-subquery expression itself. After pulling up IN-subquery expressions into a Project node, `RewritePredicateSubquery#apply` is called again to handle the `Project` as a `UnaryNode`. The `Join` will now be inserted between the `Project` and the `Aggregate` node, and the join condition will use an attribute rather than an aggregate expression, e.g.:
```
Project [col1#32, exists#42 AS (sum(col2) IN (listquery()))#40]
+- Join ExistenceJoin(exists#42), (sum(col2)#41L = c2#39L)
:- Aggregate [col1#32], [col1#32, sum(col2#33) AS sum(col2)#41L]
: +- LocalRelation [col1#32, col2#33]
+- LocalRelation [c2#39L]
```
`sum(col2)#41L` in the above join condition, despite how it looks, is the name of the attribute, not an aggregate expression.
### Why are the changes needed?
The following query fails:
```
create or replace temp view v1(c1, c2) as values (1, 2), (1, 3), (2, 2), (3, 7), (3, 1);
create or replace temp view v2(col1, col2) as values (1, 2), (1, 3), (2, 2), (3, 7), (3, 1);
select col1, sum(col2) in (select c2 from v1)
from v2 group by col1;
```
It fails with this error:
```
[INTERNAL_ERROR] Cannot generate code for expression: sum(input[1, int, false]) SQLSTATE: XX000
```
With SPARK_TESTING=1, it fails with this error:
```
[PLAN_VALIDATION_FAILED_RULE_IN_BATCH] Rule org.apache.spark.sql.catalyst.optimizer.RewritePredicateSubquery in batch RewriteSubquery generated an invalid plan: Special expressions are placed in the wrong plan:
Aggregate [col1#11], [col1#11, first(exists#20, false) AS (sum(col2) IN (listquery()))#19]
+- Join ExistenceJoin(exists#20), (sum(col2#12) = c2#18L)
:- LocalRelation [col1#11, col2#12]
+- LocalRelation [c2#18L]
```
The issue is that `RewritePredicateSubquery` builds a `Join` operator where the join condition contains an aggregate expression.
The bug is in the handler for `UnaryNode` in `RewritePredicateSubquery#apply`, which adds a `Join` below the `Aggregate` and assumes that the left-hand operand of IN-subquery can be used in the join condition. This works fine for most cases, but not when the left-hand operand is an aggregate expression.
This PR moves the offending IN-subqueries to a `Project` node, with the aggregates replaced by attributes referring to the aggregate expressions. The resulting join condition now uses those attributes rather than the actual aggregate expressions.
### Does this PR introduce _any_ user-facing change?
No, other than allowing this type of query to succeed.
### How was this patch tested?
New unit tests.
### Was this patch authored or co-authored using generative AI tooling?
No.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50091][SQL] Handle case of aggregates in left-hand operand of IN-subquery [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #48627:
URL: https://github.com/apache/spark/pull/48627#discussion_r1903631161
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -246,46 +267,242 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper {
}
}
- case u: UnaryNode if u.expressions.exists(
- SubqueryExpression.hasInOrCorrelatedExistsSubquery) =>
- var newChild = u.child
- var introducedAttrs = Seq.empty[Attribute]
- val updatedNode = u.mapExpressions(expr => {
- val (newExpr, p, newAttrs) = rewriteExistentialExprWithAttrs(Seq(expr), newChild)
- newChild = p
- introducedAttrs ++= newAttrs
- // The newExpr can not be None
- newExpr.get
- }).withNewChildren(Seq(newChild))
- updatedNode match {
- case a: Aggregate if conf.getConf(WRAP_EXISTS_IN_AGGREGATE_FUNCTION) =>
- // If we have introduced new `exists`-attributes that are referenced by
- // aggregateExpressions within a non-aggregateFunction expression, we wrap them in
- // first() aggregate function. first() is Spark's executable version of any_value()
- // aggregate function.
- // We do this to keep the aggregation valid, i.e avoid references outside of aggregate
- // functions that are not in grouping expressions.
- // Note that the same `exists` attr will never appear in groupingExpressions due to
- // PullOutGroupingExpressions rule.
- // Also note: the value of `exists` is functionally determined by grouping expressions,
- // so applying any aggregate function is semantically safe.
- val aggFunctionReferences = a.aggregateExpressions.
- flatMap(extractAggregateExpressions).
- flatMap(_.references).toSet
- val nonAggFuncReferences =
- a.aggregateExpressions.flatMap(_.references).filterNot(aggFunctionReferences.contains)
- val toBeWrappedExistsAttrs = introducedAttrs.filter(nonAggFuncReferences.contains)
-
- // Replace all eligible `exists` by `First(exists)` among aggregateExpressions.
- val newAggregateExpressions = a.aggregateExpressions.map { aggExpr =>
- aggExpr.transformUp {
- case attr: Attribute if toBeWrappedExistsAttrs.contains(attr) =>
- new First(attr).toAggregateExpression()
- }.asInstanceOf[NamedExpression]
- }
- a.copy(aggregateExpressions = newAggregateExpressions)
- case _ => updatedNode
+ // Handle the case where the left-hand side of an IN-subquery contains an aggregate.
+ //
+ // This handler pulls up any expression containing such an IN-subquery into a new Project
+ // node, replacing aggregate expressions with attributes. The new Project node will be
+ // handled by the Unary node handler.
+ //
+ // The Unary node handler uses the left-hand side of the IN-subquery in a
+ // join condition. Thus, without this pre-transformation, the join condition
+ // contains an aggregate, which is illegal. With this pre-transformation, the
+ // join condition contains an attribute from the left-hand side of the
+ // IN-subquery contained in the Project node.
+ //
+ // For example:
+ //
+ // SELECT col1, SUM(col2) IN (SELECT c2 FROM v1) as x
+ // FROM v2 GROUP BY col1;
+ //
+ // The above query has this plan on entry to RewritePredicateSubquery#apply:
+ //
+ // Aggregate [col1#28], [col1#28, sum(col2#29) IN (list#24 []) AS x#25]
+ // : +- LocalRelation [c2#35L]
+ // +- LocalRelation [col1#28, col2#29]
+ //
+ // Note that the Aggregate node contains the IN-subquery and the left-hand
+ // side of the IN-subquery is an aggregate expression (sum(col2#29)).
+ //
+ // This handler transforms the above plan into the following:
+ //
+ // Project [col1#28, sum(col2)#36L IN (list#24 []) AS x#25]
+ // : +- LocalRelation [c2#35L]
+ // +- Aggregate [col1#28], [col1#28, sum(col2#29) AS sum(col2)#36L]
+ // +- LocalRelation [col1#28, col2#29]
+ //
+ // The transformation pulled the IN-subquery up into a Project. The left-hand side of the
+ // IN-subquery is now an attribute (sum(col2)#36L) that refers to the actual aggregation
+ // which is still performed in the Aggregate node (sum(col2#29) AS sum(col2)#36L). The Unary
+ // node handler will use that attribute in the join condition (rather than the aggregate
+ // expression).
+ //
+ // If the IN-subquery is nested in a larger expression, that entire larger
+ // expression is pulled up into the Project. For example:
+ //
+ // SELECT SUM(col2) IN (SELECT c3 FROM v1) AND SUM(col3) > -1 AS x
+ // FROM v2;
+ //
+ // The input to RewritePredicateSubquery#apply is the following plan:
+ //
+ // Aggregate [(sum(col2#34) IN (list#28 []) AND (sum(col3#35) > -1)) AS x#29]
+ // : +- LocalRelation [c3#44L]
+ // +- LocalRelation [col2#34, col3#35]
+ //
+ // This handler transforms the plan into:
+ //
+ // Project [(sum(col2)#45L IN (list#28 []) AND (sum(col3)#46L > -1)) AS x#29]
+ // : +- LocalRelation [c3#44L]
+ // +- Aggregate [sum(col2#34) AS sum(col2)#45L, sum(col3#35) AS sum(col3)#46L]
+ // +- LocalRelation [col2#34, col3#35]
+ //
+ // Note that the entire AND expression was pulled up into the Project, but the Aggregate
+ // node continues to perform the aggregations (but without the IN-subquery expression).
+ case a: Aggregate if exprsContainsAggregateInSubquery(a.aggregateExpressions) =>
+ // Find any interesting expressions from Aggregate.aggregateExpressions.
+ //
+ // An interesting expression is one that contains an IN-subquery whose left-hand
+ // operand contains aggregates. For example:
+ //
+ // SELECT col1, SUM(col2) IN (SELECT c2 FROM v1)
+ // FROM v2 GROUP BY col1;
+ //
+ // withInSubquery will be a List containing a single Alias expression:
+ //
+ // List(sum(col2#12) IN (list#8 []) AS (...)#19)
+ val withInSubquery = a.aggregateExpressions.filter(exprContainsAggregateInSubquery(_))
Review Comment:
BTW I think it's better to always build the query plan tree with this normalized form (`Aggregate` should only do grouping and aggregating, projection should always happen in `Project`), but this is a much bigger topic.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50091][SQL] Handle case of aggregates in left-hand operand of IN-subquery [spark]
Posted by "bersprockets (via GitHub)" <gi...@apache.org>.
bersprockets commented on code in PR #48627:
URL: https://github.com/apache/spark/pull/48627#discussion_r1859632024
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -246,6 +267,197 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper {
}
}
+ // Handle the case where the left-hand side of an IN-subquery contains an aggregate.
+ //
+ // This handler pulls up any expression containing such an IN-subquery into a new Project
+ // node, replacing aggregate expressions with attributes, and then re-enters
+ // RewritePredicateSubquery#apply, where the new Project node will be handled
+ // by the Unary node handler.
+ //
+ // The Unary node handler uses the left-hand side of the IN-subquery in a
+ // join condition. Thus, without this pre-transformation, the join condition
+ // contains an aggregate, which is illegal. With this pre-transformation, the
+ // join condition contains an attribute from the left-hand side of the
+ // IN-subquery contained in the Project node.
+ //
+ // For example:
+ //
+ // SELECT col1, SUM(col2) IN (SELECT c2 FROM v1) as x
+ // FROM v2 GROUP BY col1;
+ //
+ // The above query has this plan on entry to RewritePredicateSubquery#apply:
+ //
+ // Aggregate [col1#28], [col1#28, sum(col2#29) IN (list#24 []) AS x#25]
+ // : +- LocalRelation [c2#35L]
+ // +- LocalRelation [col1#28, col2#29]
+ //
+ // Note that the Aggregate node contains the IN-subquery and the left-hand
+ // side of the IN-subquery is an aggregate expression (sum(col2#28)).
+ //
+ // This handler transforms the above plan into the following:
+ //
+ // Project [col1#28, sum(col2)#36L IN (list#24 []) AS x#25]
+ // : +- LocalRelation [c2#35L]
+ // +- Aggregate [col1#28], [col1#28, sum(col2#29) AS sum(col2)#36L]
+ // +- LocalRelation [col1#28, col2#29]
+ //
+ // The transformation pulled the IN-subquery up into a Project. The left-hand side of the
+ // IN-subquery is now an attribute (sum(col2)#36L) that refers to the actual aggregation
+ // which is still performed in the Aggregate node (sum(col2#28) AS sum(col2)#36L). The Unary
Review Comment:
Good catch!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50091][SQL] Handle case of aggregates in left-hand operand of IN-subquery [spark]
Posted by "bersprockets (via GitHub)" <gi...@apache.org>.
bersprockets commented on code in PR #48627:
URL: https://github.com/apache/spark/pull/48627#discussion_r1911053711
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -246,46 +267,242 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper {
}
}
- case u: UnaryNode if u.expressions.exists(
- SubqueryExpression.hasInOrCorrelatedExistsSubquery) =>
- var newChild = u.child
- var introducedAttrs = Seq.empty[Attribute]
- val updatedNode = u.mapExpressions(expr => {
- val (newExpr, p, newAttrs) = rewriteExistentialExprWithAttrs(Seq(expr), newChild)
- newChild = p
- introducedAttrs ++= newAttrs
- // The newExpr can not be None
- newExpr.get
- }).withNewChildren(Seq(newChild))
- updatedNode match {
- case a: Aggregate if conf.getConf(WRAP_EXISTS_IN_AGGREGATE_FUNCTION) =>
- // If we have introduced new `exists`-attributes that are referenced by
- // aggregateExpressions within a non-aggregateFunction expression, we wrap them in
- // first() aggregate function. first() is Spark's executable version of any_value()
- // aggregate function.
- // We do this to keep the aggregation valid, i.e avoid references outside of aggregate
- // functions that are not in grouping expressions.
- // Note that the same `exists` attr will never appear in groupingExpressions due to
- // PullOutGroupingExpressions rule.
- // Also note: the value of `exists` is functionally determined by grouping expressions,
- // so applying any aggregate function is semantically safe.
- val aggFunctionReferences = a.aggregateExpressions.
- flatMap(extractAggregateExpressions).
- flatMap(_.references).toSet
- val nonAggFuncReferences =
- a.aggregateExpressions.flatMap(_.references).filterNot(aggFunctionReferences.contains)
- val toBeWrappedExistsAttrs = introducedAttrs.filter(nonAggFuncReferences.contains)
-
- // Replace all eligible `exists` by `First(exists)` among aggregateExpressions.
- val newAggregateExpressions = a.aggregateExpressions.map { aggExpr =>
- aggExpr.transformUp {
- case attr: Attribute if toBeWrappedExistsAttrs.contains(attr) =>
- new First(attr).toAggregateExpression()
- }.asInstanceOf[NamedExpression]
- }
- a.copy(aggregateExpressions = newAggregateExpressions)
- case _ => updatedNode
+ // Handle the case where the left-hand side of an IN-subquery contains an aggregate.
+ //
+ // This handler pulls up any expression containing such an IN-subquery into a new Project
+ // node, replacing aggregate expressions with attributes. The new Project node will be
+ // handled by the Unary node handler.
+ //
+ // The Unary node handler uses the left-hand side of the IN-subquery in a
+ // join condition. Thus, without this pre-transformation, the join condition
+ // contains an aggregate, which is illegal. With this pre-transformation, the
+ // join condition contains an attribute from the left-hand side of the
+ // IN-subquery contained in the Project node.
+ //
+ // For example:
+ //
+ // SELECT col1, SUM(col2) IN (SELECT c2 FROM v1) as x
+ // FROM v2 GROUP BY col1;
+ //
+ // The above query has this plan on entry to RewritePredicateSubquery#apply:
+ //
+ // Aggregate [col1#28], [col1#28, sum(col2#29) IN (list#24 []) AS x#25]
+ // : +- LocalRelation [c2#35L]
+ // +- LocalRelation [col1#28, col2#29]
+ //
+ // Note that the Aggregate node contains the IN-subquery and the left-hand
+ // side of the IN-subquery is an aggregate expression (sum(col2#29)).
+ //
+ // This handler transforms the above plan into the following:
+ //
+ // Project [col1#28, sum(col2)#36L IN (list#24 []) AS x#25]
+ // : +- LocalRelation [c2#35L]
+ // +- Aggregate [col1#28], [col1#28, sum(col2#29) AS sum(col2)#36L]
+ // +- LocalRelation [col1#28, col2#29]
+ //
+ // The transformation pulled the IN-subquery up into a Project. The left-hand side of the
+ // IN-subquery is now an attribute (sum(col2)#36L) that refers to the actual aggregation
+ // which is still performed in the Aggregate node (sum(col2#29) AS sum(col2)#36L). The Unary
+ // node handler will use that attribute in the join condition (rather than the aggregate
+ // expression).
+ //
+ // If the IN-subquery is nested in a larger expression, that entire larger
+ // expression is pulled up into the Project. For example:
+ //
+ // SELECT SUM(col2) IN (SELECT c3 FROM v1) AND SUM(col3) > -1 AS x
+ // FROM v2;
+ //
+ // The input to RewritePredicateSubquery#apply is the following plan:
+ //
+ // Aggregate [(sum(col2#34) IN (list#28 []) AND (sum(col3#35) > -1)) AS x#29]
+ // : +- LocalRelation [c3#44L]
+ // +- LocalRelation [col2#34, col3#35]
+ //
+ // This handler transforms the plan into:
+ //
+ // Project [(sum(col2)#45L IN (list#28 []) AND (sum(col3)#46L > -1)) AS x#29]
+ // : +- LocalRelation [c3#44L]
+ // +- Aggregate [sum(col2#34) AS sum(col2)#45L, sum(col3#35) AS sum(col3)#46L]
+ // +- LocalRelation [col2#34, col3#35]
+ //
+ // Note that the entire AND expression was pulled up into the Project, but the Aggregate
+ // node continues to perform the aggregations (but without the IN-subquery expression).
+ case a: Aggregate if exprsContainsAggregateInSubquery(a.aggregateExpressions) =>
+ // Find any interesting expressions from Aggregate.aggregateExpressions.
+ //
+ // An interesting expression is one that contains an IN-subquery whose left-hand
+ // operand contains aggregates. For example:
+ //
+ // SELECT col1, SUM(col2) IN (SELECT c2 FROM v1)
+ // FROM v2 GROUP BY col1;
+ //
+ // withInSubquery will be a List containing a single Alias expression:
+ //
+ // List(sum(col2#12) IN (list#8 []) AS (...)#19)
+ val withInSubquery = a.aggregateExpressions.filter(exprContainsAggregateInSubquery(_))
Review Comment:
I am delayed in responding to review comments: I not around my laptop much until next week.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50091][SQL] Handle case of aggregates in left-hand operand of IN-subquery [spark]
Posted by "bersprockets (via GitHub)" <gi...@apache.org>.
bersprockets commented on code in PR #48627:
URL: https://github.com/apache/spark/pull/48627#discussion_r1853016997
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -245,6 +266,55 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper {
condition = Some(newCondition)))
}
}
+ case a: Aggregate if exprsContainsAggregateInSubquery(a.aggregateExpressions) =>
+ // find expressions with an IN-subquery whose left-hand operand contains aggregates
+ val withInSubquery = a.aggregateExpressions.filter(exprContainsAggregateInSubquery(_))
+
+ // extract the aggregate expressions from withInSubquery
+ val inSubqueryMapping = withInSubquery.map { e =>
+ (e, extractAggregateExpressions(e))
+ }
+
+ val inSubqueryMap = inSubqueryMapping.toMap
+ // get all aggregate expressions found in left-hand operands of IN-subqueries
+ val aggregateExprs = inSubqueryMapping.flatMap(_._2)
+ // create aliases for each above aggregate expression
+ val aggregateExprAliases = aggregateExprs.map(a => Alias(a, toPrettySQL(a))())
Review Comment:
> I have a question here: why this alias is needed?
>
`aggregateExpressions` in `Aggregate` is defined as `Seq[NamedExpression]`. `Alias` is a `NamedExpression` but `AggregateExpression` is not. `aggregateExpressions` typically contains attributes (for pass-throughs) or aliases (for aggregate expressions).
I add a comment in the code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50091][SQL] Handle case of aggregates in left-hand operand of IN-subquery [spark]
Posted by "bersprockets (via GitHub)" <gi...@apache.org>.
bersprockets commented on code in PR #48627:
URL: https://github.com/apache/spark/pull/48627#discussion_r1923076982
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteSubquerySuite.scala:
##########
@@ -79,4 +80,20 @@ class RewriteSubquerySuite extends PlanTest {
Optimize.executeAndTrack(query.analyze, tracker)
assert(tracker.rules(RewritePredicateSubquery.ruleName).numEffectiveInvocations == 0)
}
+
+ test("SPARK-50091: Don't put aggregate expression in join condition") {
Review Comment:
I also updated this test to check the whole optimized plan rather than simply testing that the join condition does not have an aggregate expression.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50091][SQL] Handle case of aggregates in left-hand operand of IN-subquery [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #48627:
URL: https://github.com/apache/spark/pull/48627#discussion_r1901465121
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -246,6 +267,197 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper {
}
}
+ // Handle the case where the left-hand side of an IN-subquery contains an aggregate.
+ //
+ // This handler pulls up any expression containing such an IN-subquery into a new Project
+ // node, replacing aggregate expressions with attributes, and then re-enters
+ // RewritePredicateSubquery#apply, where the new Project node will be handled
+ // by the Unary node handler.
+ //
+ // The Unary node handler uses the left-hand side of the IN-subquery in a
+ // join condition. Thus, without this pre-transformation, the join condition
+ // contains an aggregate, which is illegal. With this pre-transformation, the
+ // join condition contains an attribute from the left-hand side of the
+ // IN-subquery contained in the Project node.
+ //
+ // For example:
+ //
+ // SELECT col1, SUM(col2) IN (SELECT c2 FROM v1) as x
+ // FROM v2 GROUP BY col1;
+ //
+ // The above query has this plan on entry to RewritePredicateSubquery#apply:
+ //
+ // Aggregate [col1#28], [col1#28, sum(col2#29) IN (list#24 []) AS x#25]
+ // : +- LocalRelation [c2#35L]
+ // +- LocalRelation [col1#28, col2#29]
+ //
+ // Note that the Aggregate node contains the IN-subquery and the left-hand
+ // side of the IN-subquery is an aggregate expression (sum(col2#29)).
+ //
+ // This handler transforms the above plan into the following:
+ //
+ // Project [col1#28, sum(col2)#36L IN (list#24 []) AS x#25]
+ // : +- LocalRelation [c2#35L]
+ // +- Aggregate [col1#28], [col1#28, sum(col2#29) AS sum(col2)#36L]
+ // +- LocalRelation [col1#28, col2#29]
+ //
+ // The transformation pulled the IN-subquery up into a Project. The left-hand side of the
+ // IN-subquery is now an attribute (sum(col2)#36L) that refers to the actual aggregation
+ // which is still performed in the Aggregate node (sum(col2#29) AS sum(col2)#36L). The Unary
+ // node handler will use that attribute in the join condition (rather than the aggregate
+ // expression).
+ //
+ // If the IN-subquery is nested in a larger expression, that entire larger
+ // expression is pulled up into the Project. For example:
+ //
+ // SELECT SUM(col2) IN (SELECT c3 FROM v1) AND SUM(col3) > -1 AS x
+ // FROM v2;
+ //
+ // The input to RewritePredicateSubquery#apply is the following plan:
+ //
+ // Aggregate [(sum(col2#34) IN (list#28 []) AND (sum(col3#35) > -1)) AS x#29]
+ // : +- LocalRelation [c3#44L]
+ // +- LocalRelation [col2#34, col3#35]
+ //
+ // This handler transforms the plan into:
+ //
+ // Project [(sum(col2)#45L IN (list#28 []) AND (sum(col3)#46L > -1)) AS x#29]
+ // : +- LocalRelation [c3#44L]
+ // +- Aggregate [sum(col2#34) AS sum(col2)#45L, sum(col3#35) AS sum(col3)#46L]
+ // +- LocalRelation [col2#34, col3#35]
+ //
+ // Note that the entire AND expression was pulled up into the Project, but the Aggregate
+ // node continues to perform the aggregations (but without the IN-subquery expression).
+ case a: Aggregate if exprsContainsAggregateInSubquery(a.aggregateExpressions) =>
+ // Find any interesting expressions from Aggregate.aggregateExpressions.
+ //
+ // An interesting expression is one that contains an IN-subquery whose left-hand
+ // operand contains aggregates. For example:
+ //
+ // SELECT col1, SUM(col2) IN (SELECT c2 FROM v1)
+ // FROM v2 GROUP BY col1;
+ //
+ // withInSubquery will be a List containing a single Alias expression:
+ //
+ // List(sum(col2#12) IN (list#8 []) AS (...)#19)
+ val withInSubquery = a.aggregateExpressions.filter(exprContainsAggregateInSubquery(_))
+
+ // Extract the aggregate expressions from each interesting expression. This will include
+ // any aggregate expressions that were not part of the IN-subquery but were part
+ // of the larger containing expression.
+ val inSubqueryMapping = withInSubquery.map { e =>
+ (e, extractAggregateExpressions(e))
+ }
+
+ // Map each interesting expression to its contained aggregate expressions.
+ //
+ // Example #1:
+ //
+ // SELECT col1, SUM(col2) IN (SELECT c2 FROM v1)
+ // FROM v2 GROUP BY col1;
+ //
+ // inSubqueryMap will have a single entry mapping an Alias expression to a Vector
+ // with a single aggregate expression:
+ //
+ // Map(
+ // sum(col2#100) IN (list []) AS (...)#107 -> Vector(sum(col2#100))
+ // )
+ //
+ // Example #2:
+ //
+ // SELECT (SUM(col1), SUM(col2)) IN (SELECT c1, c2 FROM v1)
+ // FROM v2;
+ //
+ // inSubqueryMap will have a single entry mapping an Alias expression to a Vector
+ // with two aggregate expressions:
+ //
+ // Map(
+ // named_struct(_0, sum(col1#169), _1, sum(col2#170)) IN (list#166 []) AS (...)#179
+ // -> Vector(sum(col1#169), sum(col2#170))
+ // )
+ //
+ // Example #3:
+ //
+ // select SUM(col1) IN (SELECT c1 FROM v1), SUM(col2) IN (SELECT c2 FROM v1)
+ // FROM v2;
+ //
+ // inSubqueryMap will have two entries, each mapping an Alias expression to a Vector
+ // with a single aggregate expression:
+ //
+ // Map(
+ // sum(col1#193) IN (list#189 []) AS (...)#207 -> Vector(sum(col1#193)),
+ // sum(col2#194) IN (list#190 []) AS (...)#208 -> Vector(sum(col2#194))
+ // )
+ //
+ // Example #5:
+ //
+ // SELECT SUM(col2) IN (SELECT c3 FROM v1) AND SUM(col3) > -1 AS x
+ // FROM v2;
+ //
+ // inSubqueryMap will contain a single AND expression that maps to two aggregate
+ // expressions, even though only one of those aggregate expressions is used as
+ // the left-hand operand of the IN-subquery expression.
+ //
+ // Map(
+ // (sum(col2#34) IN (list#28 []) AND (sum(col3#35) > -1)) AS x#29
+ // -> Vector(sum(col2#34), sum(col3#35))
+ // )
+ //
+ // The keys of inSubqueryMap will be used to determine which expressions in
+ // the old Aggregate node are interesting. The values of inSubqueryMap, after
+ // being wrapped in Alias expressions, will replace their associated interesting
+ // expressions in a new Aggregate node.
+ val inSubqueryMap = inSubqueryMapping.toMap
+
+ // Get all aggregate expressions associated with interesting expressions.
+ val aggregateExprs = inSubqueryMapping.flatMap(_._2)
+ // Create aliases for each above aggregate expression. We can't use the aggregate
+ // expressions directly in the new Aggregate node because Aggregate.aggregateExpressions
+ // has the type Seq[NamedExpression].
+ val aggregateExprAliases = aggregateExprs.map(a => Alias(a, toPrettySQL(a))())
+ // Create a mapping from each aggregate expression to its alias.
+ val aggregateExprAliasMap = aggregateExprs.zip(aggregateExprAliases).toMap
+ // Create attributes from those aliases of aggregate expressions. These attributes
+ // will be used in the new Project node to refer to the aliased aggregate expressions
+ // in the new Aggregate node.
+ val aggregateExprAttrs = aggregateExprAliases.map(_.toAttribute)
+ // Create a mapping from aggregate expressions to attributes. This will be
+ // used when patching the interesting expressions after they are pulled up
+ // into the new Project node: aggregate expressions will be replaced by attributes.
+ val aggregateExprAttrMap = aggregateExprs.zip(aggregateExprAttrs).toMap
+
+ // Create an Aggregate node without the interesting expressions, just
+ // the associated aggregate expressions plus any other group-by or aggregate expressions
+ // that were not involved in the interesting expressions.
+ val newAggregateExpressions = a.aggregateExpressions.flatMap {
+ // If this expression contains IN-subqueries with aggregates in the left-hand
+ // operand, replace with just the aggregates.
+ case ae: Expression if inSubqueryMap.contains(ae) =>
+ // Replace the expression with an aliased aggregate expression.
+ inSubqueryMap(ae).map(aggregateExprAliasMap(_))
+ case ae => Seq(ae)
+ }
+ val newAggregate = a.copy(aggregateExpressions = newAggregateExpressions)
+
+ // Create a projection with the IN-subquery expressions that contain aggregates, replacing
+ // the aggregate expressions with attribute references to the output of the new Aggregate
+ // operator. Also include the other output of the Aggregate operator.
+ val projList = a.aggregateExpressions.map {
+ // If this expression contains an IN-subquery that uses an aggregate, we
+ // need to do something special
+ case ae: Expression if inSubqueryMap.contains(ae) =>
+ ae.transform {
+ // Patch any aggregate expression with its corresponding attribute.
+ case a: AggregateExpression => aggregateExprAttrMap(a)
+ }.asInstanceOf[NamedExpression]
+ case ae => ae.toAttribute
+ }
+ val newProj = Project(projList, newAggregate)
+
+ // Reapply this rule, but now with all interesting expressions
+ // from Aggregate.aggregateExpressions pulled up into a Project node.
+ apply(newProj)
Review Comment:
This reminds me of the rule `RewriteWithExpression`, which also needs to rewrite `Aggregate` first. We should not call `apply` here in the middle of plan traveral, as `apply` transforms the plan again, and leads to O(n^2) complexity. Instead of, we should also add a util function that rewrites `UnaryNode` (not transforms the full tree) and call it here and the original case match for `UnaryNode`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50091][SQL] Handle case of aggregates in left-hand operand of IN-subquery [spark]
Posted by "bersprockets (via GitHub)" <gi...@apache.org>.
bersprockets commented on code in PR #48627:
URL: https://github.com/apache/spark/pull/48627#discussion_r1855579253
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -246,6 +267,191 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper {
}
}
+ // Handle the case where the left-hand side of an IN-subquery contains an aggregate.
Review Comment:
My new comment still needs a little work. Will be posting an update.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50091][SQL] Handle case of aggregates in left-hand operand of IN-subquery [spark]
Posted by "bersprockets (via GitHub)" <gi...@apache.org>.
bersprockets commented on PR #48627:
URL: https://github.com/apache/spark/pull/48627#issuecomment-2479232015
cc @cloud-fan
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50091][SQL] Handle case of aggregates in left-hand operand of IN-subquery [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #48627:
URL: https://github.com/apache/spark/pull/48627#issuecomment-2608746953
thanks, merging to master/4.0!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50091][SQL] Handle case of aggregates in left-hand operand of IN-subquery [spark]
Posted by "bersprockets (via GitHub)" <gi...@apache.org>.
bersprockets commented on code in PR #48627:
URL: https://github.com/apache/spark/pull/48627#discussion_r1923076021
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -246,46 +267,242 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper {
}
}
- case u: UnaryNode if u.expressions.exists(
- SubqueryExpression.hasInOrCorrelatedExistsSubquery) =>
- var newChild = u.child
- var introducedAttrs = Seq.empty[Attribute]
- val updatedNode = u.mapExpressions(expr => {
- val (newExpr, p, newAttrs) = rewriteExistentialExprWithAttrs(Seq(expr), newChild)
- newChild = p
- introducedAttrs ++= newAttrs
- // The newExpr can not be None
- newExpr.get
- }).withNewChildren(Seq(newChild))
- updatedNode match {
- case a: Aggregate if conf.getConf(WRAP_EXISTS_IN_AGGREGATE_FUNCTION) =>
- // If we have introduced new `exists`-attributes that are referenced by
- // aggregateExpressions within a non-aggregateFunction expression, we wrap them in
- // first() aggregate function. first() is Spark's executable version of any_value()
- // aggregate function.
- // We do this to keep the aggregation valid, i.e avoid references outside of aggregate
- // functions that are not in grouping expressions.
- // Note that the same `exists` attr will never appear in groupingExpressions due to
- // PullOutGroupingExpressions rule.
- // Also note: the value of `exists` is functionally determined by grouping expressions,
- // so applying any aggregate function is semantically safe.
- val aggFunctionReferences = a.aggregateExpressions.
- flatMap(extractAggregateExpressions).
- flatMap(_.references).toSet
- val nonAggFuncReferences =
- a.aggregateExpressions.flatMap(_.references).filterNot(aggFunctionReferences.contains)
- val toBeWrappedExistsAttrs = introducedAttrs.filter(nonAggFuncReferences.contains)
-
- // Replace all eligible `exists` by `First(exists)` among aggregateExpressions.
- val newAggregateExpressions = a.aggregateExpressions.map { aggExpr =>
- aggExpr.transformUp {
- case attr: Attribute if toBeWrappedExistsAttrs.contains(attr) =>
- new First(attr).toAggregateExpression()
- }.asInstanceOf[NamedExpression]
- }
- a.copy(aggregateExpressions = newAggregateExpressions)
- case _ => updatedNode
+ // Handle the case where the left-hand side of an IN-subquery contains an aggregate.
+ //
+ // This handler pulls up any expression containing such an IN-subquery into a new Project
+ // node, replacing aggregate expressions with attributes. The new Project node will be
+ // handled by the Unary node handler.
+ //
+ // The Unary node handler uses the left-hand side of the IN-subquery in a
+ // join condition. Thus, without this pre-transformation, the join condition
+ // contains an aggregate, which is illegal. With this pre-transformation, the
+ // join condition contains an attribute from the left-hand side of the
+ // IN-subquery contained in the Project node.
+ //
+ // For example:
+ //
+ // SELECT col1, SUM(col2) IN (SELECT c2 FROM v1) as x
+ // FROM v2 GROUP BY col1;
+ //
+ // The above query has this plan on entry to RewritePredicateSubquery#apply:
+ //
+ // Aggregate [col1#28], [col1#28, sum(col2#29) IN (list#24 []) AS x#25]
+ // : +- LocalRelation [c2#35L]
+ // +- LocalRelation [col1#28, col2#29]
+ //
+ // Note that the Aggregate node contains the IN-subquery and the left-hand
+ // side of the IN-subquery is an aggregate expression (sum(col2#29)).
+ //
+ // This handler transforms the above plan into the following:
+ //
+ // Project [col1#28, sum(col2)#36L IN (list#24 []) AS x#25]
+ // : +- LocalRelation [c2#35L]
+ // +- Aggregate [col1#28], [col1#28, sum(col2#29) AS sum(col2)#36L]
+ // +- LocalRelation [col1#28, col2#29]
+ //
+ // The transformation pulled the IN-subquery up into a Project. The left-hand side of the
+ // IN-subquery is now an attribute (sum(col2)#36L) that refers to the actual aggregation
+ // which is still performed in the Aggregate node (sum(col2#29) AS sum(col2)#36L). The Unary
+ // node handler will use that attribute in the join condition (rather than the aggregate
+ // expression).
+ //
+ // If the IN-subquery is nested in a larger expression, that entire larger
+ // expression is pulled up into the Project. For example:
+ //
+ // SELECT SUM(col2) IN (SELECT c3 FROM v1) AND SUM(col3) > -1 AS x
+ // FROM v2;
+ //
+ // The input to RewritePredicateSubquery#apply is the following plan:
+ //
+ // Aggregate [(sum(col2#34) IN (list#28 []) AND (sum(col3#35) > -1)) AS x#29]
+ // : +- LocalRelation [c3#44L]
+ // +- LocalRelation [col2#34, col3#35]
+ //
+ // This handler transforms the plan into:
+ //
+ // Project [(sum(col2)#45L IN (list#28 []) AND (sum(col3)#46L > -1)) AS x#29]
+ // : +- LocalRelation [c3#44L]
+ // +- Aggregate [sum(col2#34) AS sum(col2)#45L, sum(col3#35) AS sum(col3)#46L]
+ // +- LocalRelation [col2#34, col3#35]
+ //
+ // Note that the entire AND expression was pulled up into the Project, but the Aggregate
+ // node continues to perform the aggregations (but without the IN-subquery expression).
+ case a: Aggregate if exprsContainsAggregateInSubquery(a.aggregateExpressions) =>
+ // Find any interesting expressions from Aggregate.aggregateExpressions.
+ //
+ // An interesting expression is one that contains an IN-subquery whose left-hand
+ // operand contains aggregates. For example:
+ //
+ // SELECT col1, SUM(col2) IN (SELECT c2 FROM v1)
+ // FROM v2 GROUP BY col1;
+ //
+ // withInSubquery will be a List containing a single Alias expression:
+ //
+ // List(sum(col2#12) IN (list#8 []) AS (...)#19)
+ val withInSubquery = a.aggregateExpressions.filter(exprContainsAggregateInSubquery(_))
Review Comment:
I didn't create a util function because the `PhysicalAggregation` extractor does almost all the heavy lifting and the version of the code in `RewriteWithExpression` called `applyInternal` on the new `Aggregate` node before making it a child of the new `Project` node.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50091][SQL] Handle case of aggregates in left-hand operand of IN-subquery [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #48627:
URL: https://github.com/apache/spark/pull/48627#issuecomment-2491230189
cc @agubichev @andylam-db
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50091][SQL] Handle case of aggregates in left-hand operand of IN-subquery [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #48627:
URL: https://github.com/apache/spark/pull/48627#discussion_r1923402967
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -246,46 +267,106 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper {
}
}
+ // Handle the case where the left-hand side of an IN-subquery contains an aggregate.
+ //
+ // If an Aggregate node contains such an IN-subquery, this handler will pull up all
+ // expressions from the Aggregate node into a new Project node. The new Project node
+ // will then be handled by the Unary node handler.
+ //
+ // The Unary node handler uses the left-hand side of the IN-subquery in a
+ // join condition. Thus, without this pre-transformation, the join condition
+ // contains an aggregate, which is illegal. With this pre-transformation, the
+ // join condition contains an attribute from the left-hand side of the
+ // IN-subquery contained in the Project node.
+ //
+ // For example:
+ //
+ // SELECT SUM(col2) IN (SELECT c3 FROM v1) AND SUM(col3) > -1 AS x
+ // FROM v2;
+ //
+ // The above query has this plan on entry to RewritePredicateSubquery#apply:
+ //
+ // Aggregate [(sum(col2#18) IN (list#12 []) AND (sum(col3#19) > -1)) AS x#13]
+ // : +- LocalRelation [c3#28L]
+ // +- LocalRelation [col2#18, col3#19]
+ //
+ // Note that the Aggregate node contains the IN-subquery and the left-hand
+ // side of the IN-subquery is an aggregate expression sum(col2#18)).
+ //
+ // This handler transforms the above plan into the following:
+ // scalastyle:off line.size.limit
+ //
+ // Project [(_aggregateexpression#20L IN (list#12 []) AND (_aggregateexpression#21L > -1)) AS x#13]
+ // : +- LocalRelation [c3#28L]
+ // +- Aggregate [sum(col2#18) AS _aggregateexpression#20L, sum(col3#19) AS _aggregateexpression#21L]
+ // +- LocalRelation [col2#18, col3#19]
+ //
+ // scalastyle:on
+ // Note that both the IN-subquery and the greater-than expressions have been
+ // pulled up into the Project node. These expressions use attributes
+ // (_aggregateexpression#20L and _aggregateexpression#21L) to refer to the aggregations
+ // which are still performed in the Aggregate node (sum(col2#18) and sum(col3#19)).
+ case p @ PhysicalAggregation(
+ groupingExpressions, aggregateExpressions, resultExpressions, child)
+ if exprsContainsAggregateInSubquery(p.expressions) =>
Review Comment:
This rewrite only pulls out subquery expressions for `Aggregate#aggregateExpressions`, not grouping expressions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50091][SQL] Handle case of aggregates in left-hand operand of IN-subquery [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on code in PR #48627:
URL: https://github.com/apache/spark/pull/48627#discussion_r1859540322
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -246,6 +267,197 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper {
}
}
+ // Handle the case where the left-hand side of an IN-subquery contains an aggregate.
+ //
+ // This handler pulls up any expression containing such an IN-subquery into a new Project
+ // node, replacing aggregate expressions with attributes, and then re-enters
+ // RewritePredicateSubquery#apply, where the new Project node will be handled
+ // by the Unary node handler.
+ //
+ // The Unary node handler uses the left-hand side of the IN-subquery in a
+ // join condition. Thus, without this pre-transformation, the join condition
+ // contains an aggregate, which is illegal. With this pre-transformation, the
+ // join condition contains an attribute from the left-hand side of the
+ // IN-subquery contained in the Project node.
+ //
+ // For example:
+ //
+ // SELECT col1, SUM(col2) IN (SELECT c2 FROM v1) as x
+ // FROM v2 GROUP BY col1;
+ //
+ // The above query has this plan on entry to RewritePredicateSubquery#apply:
+ //
+ // Aggregate [col1#28], [col1#28, sum(col2#29) IN (list#24 []) AS x#25]
+ // : +- LocalRelation [c2#35L]
+ // +- LocalRelation [col1#28, col2#29]
+ //
+ // Note that the Aggregate node contains the IN-subquery and the left-hand
+ // side of the IN-subquery is an aggregate expression (sum(col2#28)).
+ //
+ // This handler transforms the above plan into the following:
+ //
+ // Project [col1#28, sum(col2)#36L IN (list#24 []) AS x#25]
+ // : +- LocalRelation [c2#35L]
+ // +- Aggregate [col1#28], [col1#28, sum(col2#29) AS sum(col2)#36L]
+ // +- LocalRelation [col1#28, col2#29]
+ //
+ // The transformation pulled the IN-subquery up into a Project. The left-hand side of the
+ // IN-subquery is now an attribute (sum(col2)#36L) that refers to the actual aggregation
+ // which is still performed in the Aggregate node (sum(col2#28) AS sum(col2)#36L). The Unary
Review Comment:
```suggestion
// which is still performed in the Aggregate node (sum(col2#29) AS sum(col2)#36L). The Unary
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -246,6 +267,197 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper {
}
}
+ // Handle the case where the left-hand side of an IN-subquery contains an aggregate.
+ //
+ // This handler pulls up any expression containing such an IN-subquery into a new Project
+ // node, replacing aggregate expressions with attributes, and then re-enters
+ // RewritePredicateSubquery#apply, where the new Project node will be handled
+ // by the Unary node handler.
+ //
+ // The Unary node handler uses the left-hand side of the IN-subquery in a
+ // join condition. Thus, without this pre-transformation, the join condition
+ // contains an aggregate, which is illegal. With this pre-transformation, the
+ // join condition contains an attribute from the left-hand side of the
+ // IN-subquery contained in the Project node.
+ //
+ // For example:
+ //
+ // SELECT col1, SUM(col2) IN (SELECT c2 FROM v1) as x
+ // FROM v2 GROUP BY col1;
+ //
+ // The above query has this plan on entry to RewritePredicateSubquery#apply:
+ //
+ // Aggregate [col1#28], [col1#28, sum(col2#29) IN (list#24 []) AS x#25]
+ // : +- LocalRelation [c2#35L]
+ // +- LocalRelation [col1#28, col2#29]
+ //
+ // Note that the Aggregate node contains the IN-subquery and the left-hand
+ // side of the IN-subquery is an aggregate expression (sum(col2#28)).
Review Comment:
```suggestion
// side of the IN-subquery is an aggregate expression (sum(col2#29)).
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50091][SQL] Handle case of aggregates in left-hand operand of IN-subquery [spark]
Posted by "bersprockets (via GitHub)" <gi...@apache.org>.
bersprockets commented on code in PR #48627:
URL: https://github.com/apache/spark/pull/48627#discussion_r1924560656
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -246,46 +267,106 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper {
}
}
+ // Handle the case where the left-hand side of an IN-subquery contains an aggregate.
+ //
+ // If an Aggregate node contains such an IN-subquery, this handler will pull up all
+ // expressions from the Aggregate node into a new Project node. The new Project node
+ // will then be handled by the Unary node handler.
+ //
+ // The Unary node handler uses the left-hand side of the IN-subquery in a
+ // join condition. Thus, without this pre-transformation, the join condition
+ // contains an aggregate, which is illegal. With this pre-transformation, the
+ // join condition contains an attribute from the left-hand side of the
+ // IN-subquery contained in the Project node.
+ //
+ // For example:
+ //
+ // SELECT SUM(col2) IN (SELECT c3 FROM v1) AND SUM(col3) > -1 AS x
+ // FROM v2;
+ //
+ // The above query has this plan on entry to RewritePredicateSubquery#apply:
+ //
+ // Aggregate [(sum(col2#18) IN (list#12 []) AND (sum(col3#19) > -1)) AS x#13]
+ // : +- LocalRelation [c3#28L]
+ // +- LocalRelation [col2#18, col3#19]
+ //
+ // Note that the Aggregate node contains the IN-subquery and the left-hand
+ // side of the IN-subquery is an aggregate expression sum(col2#18)).
+ //
+ // This handler transforms the above plan into the following:
+ // scalastyle:off line.size.limit
+ //
+ // Project [(_aggregateexpression#20L IN (list#12 []) AND (_aggregateexpression#21L > -1)) AS x#13]
+ // : +- LocalRelation [c3#28L]
+ // +- Aggregate [sum(col2#18) AS _aggregateexpression#20L, sum(col3#19) AS _aggregateexpression#21L]
+ // +- LocalRelation [col2#18, col3#19]
+ //
+ // scalastyle:on
+ // Note that both the IN-subquery and the greater-than expressions have been
+ // pulled up into the Project node. These expressions use attributes
+ // (_aggregateexpression#20L and _aggregateexpression#21L) to refer to the aggregations
+ // which are still performed in the Aggregate node (sum(col2#18) and sum(col3#19)).
+ case p @ PhysicalAggregation(
+ groupingExpressions, aggregateExpressions, resultExpressions, child)
+ if exprsContainsAggregateInSubquery(p.expressions) =>
Review Comment:
Re: `if exprsContainsAggregateInSubquery(resultExpressions) =>`.
That won't work with`exprsContainsAggregateInSubquery` as it currently stands, since that function looks for in-subqueries with aggregate expressions in the left-hand operand. `resultExpressions` has the aggregate expressions replaced with attributes, so `exprsContainsAggregateInSubquery` would never trigger.
Alternatively, I could do
```
if exprsContainsAggregateInSubquery(p.asInstanceOf[Aggregate].aggregateExpressions) =>
```
which is kind of ugly, but does the trick.
Another alternative: I'm the only one calling `exprsContainsAggregateInSubquery`, so I could change it to return true if there are any in-subqueries at all with no regard to characteristics of the left-hand operand. We would end up rewriting some cases that wouldn't otherwise cause trouble.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50091][SQL] Handle case of aggregates in left-hand operand of IN-subquery [spark]
Posted by "bersprockets (via GitHub)" <gi...@apache.org>.
bersprockets commented on code in PR #48627:
URL: https://github.com/apache/spark/pull/48627#discussion_r1855538615
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -245,6 +266,55 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper {
condition = Some(newCondition)))
}
}
+ case a: Aggregate if exprsContainsAggregateInSubquery(a.aggregateExpressions) =>
Review Comment:
@dtenedor
I could move the entire new handler into a helper function in another file.
On the other hand, this file contains 6 rules, all related to subqueries in one way or another. They could be split up (in a separate refactor, not by this PR).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50091][SQL] Handle case of aggregates in left-hand operand of IN-subquery [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #48627:
URL: https://github.com/apache/spark/pull/48627#discussion_r1903630452
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -246,46 +267,242 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper {
}
}
- case u: UnaryNode if u.expressions.exists(
- SubqueryExpression.hasInOrCorrelatedExistsSubquery) =>
- var newChild = u.child
- var introducedAttrs = Seq.empty[Attribute]
- val updatedNode = u.mapExpressions(expr => {
- val (newExpr, p, newAttrs) = rewriteExistentialExprWithAttrs(Seq(expr), newChild)
- newChild = p
- introducedAttrs ++= newAttrs
- // The newExpr can not be None
- newExpr.get
- }).withNewChildren(Seq(newChild))
- updatedNode match {
- case a: Aggregate if conf.getConf(WRAP_EXISTS_IN_AGGREGATE_FUNCTION) =>
- // If we have introduced new `exists`-attributes that are referenced by
- // aggregateExpressions within a non-aggregateFunction expression, we wrap them in
- // first() aggregate function. first() is Spark's executable version of any_value()
- // aggregate function.
- // We do this to keep the aggregation valid, i.e avoid references outside of aggregate
- // functions that are not in grouping expressions.
- // Note that the same `exists` attr will never appear in groupingExpressions due to
- // PullOutGroupingExpressions rule.
- // Also note: the value of `exists` is functionally determined by grouping expressions,
- // so applying any aggregate function is semantically safe.
- val aggFunctionReferences = a.aggregateExpressions.
- flatMap(extractAggregateExpressions).
- flatMap(_.references).toSet
- val nonAggFuncReferences =
- a.aggregateExpressions.flatMap(_.references).filterNot(aggFunctionReferences.contains)
- val toBeWrappedExistsAttrs = introducedAttrs.filter(nonAggFuncReferences.contains)
-
- // Replace all eligible `exists` by `First(exists)` among aggregateExpressions.
- val newAggregateExpressions = a.aggregateExpressions.map { aggExpr =>
- aggExpr.transformUp {
- case attr: Attribute if toBeWrappedExistsAttrs.contains(attr) =>
- new First(attr).toAggregateExpression()
- }.asInstanceOf[NamedExpression]
- }
- a.copy(aggregateExpressions = newAggregateExpressions)
- case _ => updatedNode
+ // Handle the case where the left-hand side of an IN-subquery contains an aggregate.
+ //
+ // This handler pulls up any expression containing such an IN-subquery into a new Project
+ // node, replacing aggregate expressions with attributes. The new Project node will be
+ // handled by the Unary node handler.
+ //
+ // The Unary node handler uses the left-hand side of the IN-subquery in a
+ // join condition. Thus, without this pre-transformation, the join condition
+ // contains an aggregate, which is illegal. With this pre-transformation, the
+ // join condition contains an attribute from the left-hand side of the
+ // IN-subquery contained in the Project node.
+ //
+ // For example:
+ //
+ // SELECT col1, SUM(col2) IN (SELECT c2 FROM v1) as x
+ // FROM v2 GROUP BY col1;
+ //
+ // The above query has this plan on entry to RewritePredicateSubquery#apply:
+ //
+ // Aggregate [col1#28], [col1#28, sum(col2#29) IN (list#24 []) AS x#25]
+ // : +- LocalRelation [c2#35L]
+ // +- LocalRelation [col1#28, col2#29]
+ //
+ // Note that the Aggregate node contains the IN-subquery and the left-hand
+ // side of the IN-subquery is an aggregate expression (sum(col2#29)).
+ //
+ // This handler transforms the above plan into the following:
+ //
+ // Project [col1#28, sum(col2)#36L IN (list#24 []) AS x#25]
+ // : +- LocalRelation [c2#35L]
+ // +- Aggregate [col1#28], [col1#28, sum(col2#29) AS sum(col2)#36L]
+ // +- LocalRelation [col1#28, col2#29]
+ //
+ // The transformation pulled the IN-subquery up into a Project. The left-hand side of the
+ // IN-subquery is now an attribute (sum(col2)#36L) that refers to the actual aggregation
+ // which is still performed in the Aggregate node (sum(col2#29) AS sum(col2)#36L). The Unary
+ // node handler will use that attribute in the join condition (rather than the aggregate
+ // expression).
+ //
+ // If the IN-subquery is nested in a larger expression, that entire larger
+ // expression is pulled up into the Project. For example:
+ //
+ // SELECT SUM(col2) IN (SELECT c3 FROM v1) AND SUM(col3) > -1 AS x
+ // FROM v2;
+ //
+ // The input to RewritePredicateSubquery#apply is the following plan:
+ //
+ // Aggregate [(sum(col2#34) IN (list#28 []) AND (sum(col3#35) > -1)) AS x#29]
+ // : +- LocalRelation [c3#44L]
+ // +- LocalRelation [col2#34, col3#35]
+ //
+ // This handler transforms the plan into:
+ //
+ // Project [(sum(col2)#45L IN (list#28 []) AND (sum(col3)#46L > -1)) AS x#29]
+ // : +- LocalRelation [c3#44L]
+ // +- Aggregate [sum(col2#34) AS sum(col2)#45L, sum(col3#35) AS sum(col3)#46L]
+ // +- LocalRelation [col2#34, col3#35]
+ //
+ // Note that the entire AND expression was pulled up into the Project, but the Aggregate
+ // node continues to perform the aggregations (but without the IN-subquery expression).
+ case a: Aggregate if exprsContainsAggregateInSubquery(a.aggregateExpressions) =>
+ // Find any interesting expressions from Aggregate.aggregateExpressions.
+ //
+ // An interesting expression is one that contains an IN-subquery whose left-hand
+ // operand contains aggregates. For example:
+ //
+ // SELECT col1, SUM(col2) IN (SELECT c2 FROM v1)
+ // FROM v2 GROUP BY col1;
+ //
+ // withInSubquery will be a List containing a single Alias expression:
+ //
+ // List(sum(col2#12) IN (list#8 []) AS (...)#19)
+ val withInSubquery = a.aggregateExpressions.filter(exprContainsAggregateInSubquery(_))
Review Comment:
Once we detect such `InSubquery`, I think it's much simpler to normalize the `Aggregate` node to pull up the full result projection to a new `Project` node, instead of only rewriting the problematic `InSubquery`. This is also how `RewriteWithExpression` does it and the code is much simpler and less error-prone. We can even create a util function to reuse the code in `RewriteWithExpression`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50091][SQL] Handle case of aggregates in left-hand operand of IN-subquery [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #48627:
URL: https://github.com/apache/spark/pull/48627#discussion_r1923409502
##########
sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala:
##########
@@ -2800,4 +2800,32 @@ class SubquerySuite extends QueryTest
checkAnswer(df3, Row(7))
}
}
+
+ test("SPARK-50091: Handle aggregates in left-hand operand of IN-subquery") {
+ withTable("v1", "v2") {
+ sql("""CREATE OR REPLACE TEMP VIEW v1 (c1, c2, c3) AS VALUES
+ |(1, 2, 2), (1, 5, 3), (2, 0, 4), (3, 7, 7), (3, 8, 8)""".stripMargin)
Review Comment:
nit: `Seq((1, 2, 2), (1, 5, 3), ...).toDF("c1", "c2", "c3").createTempView`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50091][SQL] Handle case of aggregates in left-hand operand of IN-subquery [spark]
Posted by "bersprockets (via GitHub)" <gi...@apache.org>.
bersprockets commented on code in PR #48627:
URL: https://github.com/apache/spark/pull/48627#discussion_r1853016997
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -245,6 +266,55 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper {
condition = Some(newCondition)))
}
}
+ case a: Aggregate if exprsContainsAggregateInSubquery(a.aggregateExpressions) =>
+ // find expressions with an IN-subquery whose left-hand operand contains aggregates
+ val withInSubquery = a.aggregateExpressions.filter(exprContainsAggregateInSubquery(_))
+
+ // extract the aggregate expressions from withInSubquery
+ val inSubqueryMapping = withInSubquery.map { e =>
+ (e, extractAggregateExpressions(e))
+ }
+
+ val inSubqueryMap = inSubqueryMapping.toMap
+ // get all aggregate expressions found in left-hand operands of IN-subqueries
+ val aggregateExprs = inSubqueryMapping.flatMap(_._2)
+ // create aliases for each above aggregate expression
+ val aggregateExprAliases = aggregateExprs.map(a => Alias(a, toPrettySQL(a))())
Review Comment:
> I have a question here: why this alias is needed?
>
`aggregateExpressions` in `Aggregate` is defined as `Seq[NamedExpression]`. `Alias` is a `NamedExpression` but `AggregateExpression` is not. `aggregateExpressions` typically contains attributes (for pass-throughs) or aliases (for aggregate expressions).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50091][SQL] Handle case of aggregates in left-hand operand of IN-subquery [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #48627:
URL: https://github.com/apache/spark/pull/48627#issuecomment-2608748921
@bersprockets feel free to open a 3.5 backport if it's also an issue there.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50091][SQL] Handle case of aggregates in left-hand operand of IN-subquery [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #48627:
URL: https://github.com/apache/spark/pull/48627#discussion_r1923402160
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -246,46 +267,106 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper {
}
}
+ // Handle the case where the left-hand side of an IN-subquery contains an aggregate.
+ //
+ // If an Aggregate node contains such an IN-subquery, this handler will pull up all
+ // expressions from the Aggregate node into a new Project node. The new Project node
+ // will then be handled by the Unary node handler.
+ //
+ // The Unary node handler uses the left-hand side of the IN-subquery in a
+ // join condition. Thus, without this pre-transformation, the join condition
+ // contains an aggregate, which is illegal. With this pre-transformation, the
+ // join condition contains an attribute from the left-hand side of the
+ // IN-subquery contained in the Project node.
+ //
+ // For example:
+ //
+ // SELECT SUM(col2) IN (SELECT c3 FROM v1) AND SUM(col3) > -1 AS x
+ // FROM v2;
+ //
+ // The above query has this plan on entry to RewritePredicateSubquery#apply:
+ //
+ // Aggregate [(sum(col2#18) IN (list#12 []) AND (sum(col3#19) > -1)) AS x#13]
+ // : +- LocalRelation [c3#28L]
+ // +- LocalRelation [col2#18, col3#19]
+ //
+ // Note that the Aggregate node contains the IN-subquery and the left-hand
+ // side of the IN-subquery is an aggregate expression sum(col2#18)).
+ //
+ // This handler transforms the above plan into the following:
+ // scalastyle:off line.size.limit
+ //
+ // Project [(_aggregateexpression#20L IN (list#12 []) AND (_aggregateexpression#21L > -1)) AS x#13]
+ // : +- LocalRelation [c3#28L]
+ // +- Aggregate [sum(col2#18) AS _aggregateexpression#20L, sum(col3#19) AS _aggregateexpression#21L]
+ // +- LocalRelation [col2#18, col3#19]
+ //
+ // scalastyle:on
+ // Note that both the IN-subquery and the greater-than expressions have been
+ // pulled up into the Project node. These expressions use attributes
+ // (_aggregateexpression#20L and _aggregateexpression#21L) to refer to the aggregations
+ // which are still performed in the Aggregate node (sum(col2#18) and sum(col3#19)).
+ case p @ PhysicalAggregation(
+ groupingExpressions, aggregateExpressions, resultExpressions, child)
+ if exprsContainsAggregateInSubquery(p.expressions) =>
Review Comment:
```suggestion
if exprsContainsAggregateInSubquery(resultExpressions) =>
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50091][SQL] Handle case of aggregates in left-hand operand of IN-subquery [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #48627:
URL: https://github.com/apache/spark/pull/48627#discussion_r1924725543
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -246,46 +267,106 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper {
}
}
+ // Handle the case where the left-hand side of an IN-subquery contains an aggregate.
+ //
+ // If an Aggregate node contains such an IN-subquery, this handler will pull up all
+ // expressions from the Aggregate node into a new Project node. The new Project node
+ // will then be handled by the Unary node handler.
+ //
+ // The Unary node handler uses the left-hand side of the IN-subquery in a
+ // join condition. Thus, without this pre-transformation, the join condition
+ // contains an aggregate, which is illegal. With this pre-transformation, the
+ // join condition contains an attribute from the left-hand side of the
+ // IN-subquery contained in the Project node.
+ //
+ // For example:
+ //
+ // SELECT SUM(col2) IN (SELECT c3 FROM v1) AND SUM(col3) > -1 AS x
+ // FROM v2;
+ //
+ // The above query has this plan on entry to RewritePredicateSubquery#apply:
+ //
+ // Aggregate [(sum(col2#18) IN (list#12 []) AND (sum(col3#19) > -1)) AS x#13]
+ // : +- LocalRelation [c3#28L]
+ // +- LocalRelation [col2#18, col3#19]
+ //
+ // Note that the Aggregate node contains the IN-subquery and the left-hand
+ // side of the IN-subquery is an aggregate expression sum(col2#18)).
+ //
+ // This handler transforms the above plan into the following:
+ // scalastyle:off line.size.limit
+ //
+ // Project [(_aggregateexpression#20L IN (list#12 []) AND (_aggregateexpression#21L > -1)) AS x#13]
+ // : +- LocalRelation [c3#28L]
+ // +- Aggregate [sum(col2#18) AS _aggregateexpression#20L, sum(col3#19) AS _aggregateexpression#21L]
+ // +- LocalRelation [col2#18, col3#19]
+ //
+ // scalastyle:on
+ // Note that both the IN-subquery and the greater-than expressions have been
+ // pulled up into the Project node. These expressions use attributes
+ // (_aggregateexpression#20L and _aggregateexpression#21L) to refer to the aggregations
+ // which are still performed in the Aggregate node (sum(col2#18) and sum(col3#19)).
+ case p @ PhysicalAggregation(
+ groupingExpressions, aggregateExpressions, resultExpressions, child)
+ if exprsContainsAggregateInSubquery(p.expressions) =>
Review Comment:
ah OK, let's keep it as it is
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50091][SQL] Handle case of aggregates in left-hand operand of IN-subquery [spark]
Posted by "dtenedor (via GitHub)" <gi...@apache.org>.
dtenedor commented on code in PR #48627:
URL: https://github.com/apache/spark/pull/48627#discussion_r1851150355
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -245,6 +266,55 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper {
condition = Some(newCondition)))
}
}
+ case a: Aggregate if exprsContainsAggregateInSubquery(a.aggregateExpressions) =>
+ // find expressions with an IN-subquery whose left-hand operand contains aggregates
+ val withInSubquery = a.aggregateExpressions.filter(exprContainsAggregateInSubquery(_))
+
+ // extract the aggregate expressions from withInSubquery
+ val inSubqueryMapping = withInSubquery.map { e =>
+ (e, extractAggregateExpressions(e))
+ }
+
+ val inSubqueryMap = inSubqueryMapping.toMap
+ // get all aggregate expressions found in left-hand operands of IN-subqueries
+ val aggregateExprs = inSubqueryMapping.flatMap(_._2)
+ // create aliases for each above aggregate expression
+ val aggregateExprAliases = aggregateExprs.map(a => Alias(a, toPrettySQL(a))())
+ // create a mapping from each aggregate expression to its alias
+ val aggregateExprAliasMap = aggregateExprs.zip(aggregateExprAliases).toMap
+ // create attributes from those aliases of aggregate expressions
+ val aggregateExprAttrs = aggregateExprAliases.map(_.toAttribute)
+ // create a mapping from aggregate expressions to attributes
+ val aggregateExprAttrMap = aggregateExprs.zip(aggregateExprAttrs).toMap
+
+ // create an Aggregate node without the offending IN-subqueries, just
+ // the aggregates themselves and all the other aggregate expressions.
+ val newAggregateExpressions = a.aggregateExpressions.flatMap {
+ // if this expression contains IN-subqueries with aggregates in the left-hand
+ // operand, replace with just the aggregates
+ case ae: Expression if inSubqueryMap.contains(ae) =>
+ // replace the expression with an aliased aggregate expression
+ inSubqueryMap(ae).map(aggregateExprAliasMap(_))
+ case ae @ _ => Seq(ae)
Review Comment:
```suggestion
case ae => Seq(ae)
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -245,6 +266,55 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper {
condition = Some(newCondition)))
}
}
+ case a: Aggregate if exprsContainsAggregateInSubquery(a.aggregateExpressions) =>
+ // find expressions with an IN-subquery whose left-hand operand contains aggregates
+ val withInSubquery = a.aggregateExpressions.filter(exprContainsAggregateInSubquery(_))
+
+ // extract the aggregate expressions from withInSubquery
+ val inSubqueryMapping = withInSubquery.map { e =>
+ (e, extractAggregateExpressions(e))
+ }
+
+ val inSubqueryMap = inSubqueryMapping.toMap
+ // get all aggregate expressions found in left-hand operands of IN-subqueries
+ val aggregateExprs = inSubqueryMapping.flatMap(_._2)
+ // create aliases for each above aggregate expression
+ val aggregateExprAliases = aggregateExprs.map(a => Alias(a, toPrettySQL(a))())
+ // create a mapping from each aggregate expression to its alias
+ val aggregateExprAliasMap = aggregateExprs.zip(aggregateExprAliases).toMap
+ // create attributes from those aliases of aggregate expressions
+ val aggregateExprAttrs = aggregateExprAliases.map(_.toAttribute)
+ // create a mapping from aggregate expressions to attributes
+ val aggregateExprAttrMap = aggregateExprs.zip(aggregateExprAttrs).toMap
+
+ // create an Aggregate node without the offending IN-subqueries, just
+ // the aggregates themselves and all the other aggregate expressions.
+ val newAggregateExpressions = a.aggregateExpressions.flatMap {
+ // if this expression contains IN-subqueries with aggregates in the left-hand
+ // operand, replace with just the aggregates
+ case ae: Expression if inSubqueryMap.contains(ae) =>
+ // replace the expression with an aliased aggregate expression
+ inSubqueryMap(ae).map(aggregateExprAliasMap(_))
+ case ae @ _ => Seq(ae)
+ }
+ val newAggregate = a.copy(aggregateExpressions = newAggregateExpressions)
+
+ // Create a projection with the IN-subquery expressions that contain aggregates, replacing
+ // the aggregate expressions with attribute references to the output of the Aggregate
+ // operator. Also include the other output of the Aggregate operator.
+ val projList = a.aggregateExpressions.map {
+ // if this expression contains an IN-subquery that uses an aggregate, we
+ // need to do something special
+ case ae: Expression if inSubqueryMap.contains(ae) =>
+ ae.transform {
+ // patch any aggregate expression with its corresponding attribute
+ case a: AggregateExpression => aggregateExprAttrMap(a)
+ }.asInstanceOf[NamedExpression]
+ case ae @ _ => ae.toAttribute
Review Comment:
```suggestion
case ae => ae.toAttribute
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -245,6 +266,55 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper {
condition = Some(newCondition)))
}
}
+ case a: Aggregate if exprsContainsAggregateInSubquery(a.aggregateExpressions) =>
+ // find expressions with an IN-subquery whose left-hand operand contains aggregates
+ val withInSubquery = a.aggregateExpressions.filter(exprContainsAggregateInSubquery(_))
+
+ // extract the aggregate expressions from withInSubquery
+ val inSubqueryMapping = withInSubquery.map { e =>
+ (e, extractAggregateExpressions(e))
+ }
+
+ val inSubqueryMap = inSubqueryMapping.toMap
+ // get all aggregate expressions found in left-hand operands of IN-subqueries
Review Comment:
It's a bit hard to follow this logic in the code. Can you add a comment with a brief example, showing the query plan and the steps performed here?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -245,6 +266,55 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper {
condition = Some(newCondition)))
}
}
+ case a: Aggregate if exprsContainsAggregateInSubquery(a.aggregateExpressions) =>
+ // find expressions with an IN-subquery whose left-hand operand contains aggregates
Review Comment:
please express the comments as full sentences (imperative is OK) starting with capital letters and ending in punctuation.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -245,6 +266,55 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper {
condition = Some(newCondition)))
}
}
+ case a: Aggregate if exprsContainsAggregateInSubquery(a.aggregateExpressions) =>
Review Comment:
This file is already over 1000 lines long, can we move this logic to a helper object in another file to improve the code health?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50091][SQL] Handle case of aggregates in left-hand operand of IN-subquery [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #48627:
URL: https://github.com/apache/spark/pull/48627#discussion_r1903631161
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -246,46 +267,242 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper {
}
}
- case u: UnaryNode if u.expressions.exists(
- SubqueryExpression.hasInOrCorrelatedExistsSubquery) =>
- var newChild = u.child
- var introducedAttrs = Seq.empty[Attribute]
- val updatedNode = u.mapExpressions(expr => {
- val (newExpr, p, newAttrs) = rewriteExistentialExprWithAttrs(Seq(expr), newChild)
- newChild = p
- introducedAttrs ++= newAttrs
- // The newExpr can not be None
- newExpr.get
- }).withNewChildren(Seq(newChild))
- updatedNode match {
- case a: Aggregate if conf.getConf(WRAP_EXISTS_IN_AGGREGATE_FUNCTION) =>
- // If we have introduced new `exists`-attributes that are referenced by
- // aggregateExpressions within a non-aggregateFunction expression, we wrap them in
- // first() aggregate function. first() is Spark's executable version of any_value()
- // aggregate function.
- // We do this to keep the aggregation valid, i.e avoid references outside of aggregate
- // functions that are not in grouping expressions.
- // Note that the same `exists` attr will never appear in groupingExpressions due to
- // PullOutGroupingExpressions rule.
- // Also note: the value of `exists` is functionally determined by grouping expressions,
- // so applying any aggregate function is semantically safe.
- val aggFunctionReferences = a.aggregateExpressions.
- flatMap(extractAggregateExpressions).
- flatMap(_.references).toSet
- val nonAggFuncReferences =
- a.aggregateExpressions.flatMap(_.references).filterNot(aggFunctionReferences.contains)
- val toBeWrappedExistsAttrs = introducedAttrs.filter(nonAggFuncReferences.contains)
-
- // Replace all eligible `exists` by `First(exists)` among aggregateExpressions.
- val newAggregateExpressions = a.aggregateExpressions.map { aggExpr =>
- aggExpr.transformUp {
- case attr: Attribute if toBeWrappedExistsAttrs.contains(attr) =>
- new First(attr).toAggregateExpression()
- }.asInstanceOf[NamedExpression]
- }
- a.copy(aggregateExpressions = newAggregateExpressions)
- case _ => updatedNode
+ // Handle the case where the left-hand side of an IN-subquery contains an aggregate.
+ //
+ // This handler pulls up any expression containing such an IN-subquery into a new Project
+ // node, replacing aggregate expressions with attributes. The new Project node will be
+ // handled by the Unary node handler.
+ //
+ // The Unary node handler uses the left-hand side of the IN-subquery in a
+ // join condition. Thus, without this pre-transformation, the join condition
+ // contains an aggregate, which is illegal. With this pre-transformation, the
+ // join condition contains an attribute from the left-hand side of the
+ // IN-subquery contained in the Project node.
+ //
+ // For example:
+ //
+ // SELECT col1, SUM(col2) IN (SELECT c2 FROM v1) as x
+ // FROM v2 GROUP BY col1;
+ //
+ // The above query has this plan on entry to RewritePredicateSubquery#apply:
+ //
+ // Aggregate [col1#28], [col1#28, sum(col2#29) IN (list#24 []) AS x#25]
+ // : +- LocalRelation [c2#35L]
+ // +- LocalRelation [col1#28, col2#29]
+ //
+ // Note that the Aggregate node contains the IN-subquery and the left-hand
+ // side of the IN-subquery is an aggregate expression (sum(col2#29)).
+ //
+ // This handler transforms the above plan into the following:
+ //
+ // Project [col1#28, sum(col2)#36L IN (list#24 []) AS x#25]
+ // : +- LocalRelation [c2#35L]
+ // +- Aggregate [col1#28], [col1#28, sum(col2#29) AS sum(col2)#36L]
+ // +- LocalRelation [col1#28, col2#29]
+ //
+ // The transformation pulled the IN-subquery up into a Project. The left-hand side of the
+ // IN-subquery is now an attribute (sum(col2)#36L) that refers to the actual aggregation
+ // which is still performed in the Aggregate node (sum(col2#29) AS sum(col2)#36L). The Unary
+ // node handler will use that attribute in the join condition (rather than the aggregate
+ // expression).
+ //
+ // If the IN-subquery is nested in a larger expression, that entire larger
+ // expression is pulled up into the Project. For example:
+ //
+ // SELECT SUM(col2) IN (SELECT c3 FROM v1) AND SUM(col3) > -1 AS x
+ // FROM v2;
+ //
+ // The input to RewritePredicateSubquery#apply is the following plan:
+ //
+ // Aggregate [(sum(col2#34) IN (list#28 []) AND (sum(col3#35) > -1)) AS x#29]
+ // : +- LocalRelation [c3#44L]
+ // +- LocalRelation [col2#34, col3#35]
+ //
+ // This handler transforms the plan into:
+ //
+ // Project [(sum(col2)#45L IN (list#28 []) AND (sum(col3)#46L > -1)) AS x#29]
+ // : +- LocalRelation [c3#44L]
+ // +- Aggregate [sum(col2#34) AS sum(col2)#45L, sum(col3#35) AS sum(col3)#46L]
+ // +- LocalRelation [col2#34, col3#35]
+ //
+ // Note that the entire AND expression was pulled up into the Project, but the Aggregate
+ // node continues to perform the aggregations (but without the IN-subquery expression).
+ case a: Aggregate if exprsContainsAggregateInSubquery(a.aggregateExpressions) =>
+ // Find any interesting expressions from Aggregate.aggregateExpressions.
+ //
+ // An interesting expression is one that contains an IN-subquery whose left-hand
+ // operand contains aggregates. For example:
+ //
+ // SELECT col1, SUM(col2) IN (SELECT c2 FROM v1)
+ // FROM v2 GROUP BY col1;
+ //
+ // withInSubquery will be a List containing a single Alias expression:
+ //
+ // List(sum(col2#12) IN (list#8 []) AS (...)#19)
+ val withInSubquery = a.aggregateExpressions.filter(exprContainsAggregateInSubquery(_))
Review Comment:
BTW I think it's better to always build the query plan tree with this normalized form (`Aggregate` should only do grouping and aggregating, projection should leave to `Project`), but this is a much bigger topic.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50091][SQL] Handle case of aggregates in left-hand operand of IN-subquery [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on code in PR #48627:
URL: https://github.com/apache/spark/pull/48627#discussion_r1852972513
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -245,6 +266,55 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper {
condition = Some(newCondition)))
}
}
+ case a: Aggregate if exprsContainsAggregateInSubquery(a.aggregateExpressions) =>
+ // find expressions with an IN-subquery whose left-hand operand contains aggregates
+ val withInSubquery = a.aggregateExpressions.filter(exprContainsAggregateInSubquery(_))
+
+ // extract the aggregate expressions from withInSubquery
+ val inSubqueryMapping = withInSubquery.map { e =>
+ (e, extractAggregateExpressions(e))
+ }
+
+ val inSubqueryMap = inSubqueryMapping.toMap
+ // get all aggregate expressions found in left-hand operands of IN-subqueries
+ val aggregateExprs = inSubqueryMapping.flatMap(_._2)
+ // create aliases for each above aggregate expression
+ val aggregateExprAliases = aggregateExprs.map(a => Alias(a, toPrettySQL(a))())
Review Comment:
I have a question here: why this alias is needed?
So instead of this:
```
!: +- Project [cast(c3#19 as bigint) AS c3#25L] +- Join ExistenceJoin(exists#28), (sum(col2)#27L = c3#25L)
!: +- Project [c3#19] :- Aggregate [sum(col2#21) AS sum(col2)#27L]
```
Can we just keep the old expression, like this:
```
!: +- Project [cast(c3#19 as bigint) AS c3#25L] +- Join ExistenceJoin(exists#28), (sum(col2#21) = c3#25L)
!: +- Project [c3#19] :- Aggregate [sum(col2#21)]
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50091][SQL] Handle case of aggregates in left-hand operand of IN-subquery [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #48627: [SPARK-50091][SQL] Handle case of aggregates in left-hand operand of IN-subquery
URL: https://github.com/apache/spark/pull/48627
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50091][SQL] Handle case of aggregates in left-hand operand of IN-subquery [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #48627:
URL: https://github.com/apache/spark/pull/48627#discussion_r1923408425
##########
sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala:
##########
@@ -2800,4 +2800,32 @@ class SubquerySuite extends QueryTest
checkAnswer(df3, Row(7))
}
}
+
+ test("SPARK-50091: Handle aggregates in left-hand operand of IN-subquery") {
+ withTable("v1", "v2") {
Review Comment:
```suggestion
withTempView("v1", "v2") {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org