This is an automated email from the ASF dual-hosted git repository.
gurwls223 pushed a commit to branch branch-3.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.2 by this push:
new 5976a06 [SPARK-38221][SQL] Eagerly iterate over groupingExpressions
when moving complex grouping expressions out of an Aggregate node
5976a06 is described below
commit 5976a0655e78b1d5666d8b2b5d459eafc6319547
Author: Bruce Robbins <[email protected]>
AuthorDate: Wed Feb 16 11:12:33 2022 +0900
[SPARK-38221][SQL] Eagerly iterate over groupingExpressions when moving
complex grouping expressions out of an Aggregate node
### What changes were proposed in this pull request?
Change `PullOutGroupingExpressions` to eagerly iterate over
`groupingExpressions` when building `complexGroupingExpressionMap`.
### Why are the changes needed?
Consider this query:
```
Seq(1).toDF("id").groupBy(Stream($"id" + 1, $"id" + 2):
_*).sum("id").show(false)
```
It fails with
```
java.lang.IllegalStateException: Couldn't find _groupingexpression#24 in
[id#4,_groupingexpression#23]
at
org.apache.spark.sql.catalyst.expressions.BindReferences$$anonfun$bindReference$1.applyOrElse(BoundAttribute.scala:80)
at
org.apache.spark.sql.catalyst.expressions.BindReferences$$anonfun$bindReference$1.applyOrElse(BoundAttribute.scala:73)
at
org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDownWithPruning$1(TreeNode.scala:481)
at
org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:83)
at
org.apache.spark.sql.catalyst.trees.TreeNode.transformDownWithPruning(TreeNode.scala:481)
at
org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:457)
at
org.apache.spark.sql.catalyst.trees.TreeNode.transform(TreeNode.scala:425)
at
org.apache.spark.sql.catalyst.expressions.BindReferences$.bindReference(BoundAttribute.scala:73)
at
org.apache.spark.sql.catalyst.expressions.BindReferences$.$anonfun$bindReferences$1(BoundAttribute.scala:94)
at scala.collection.immutable.Stream.$anonfun$map$1(Stream.scala:418)
at scala.collection.immutable.Stream$Cons.tail(Stream.scala:1173)
at scala.collection.immutable.Stream$Cons.tail(Stream.scala:1163)
at scala.collection.immutable.Stream.$anonfun$map$1(Stream.scala:418)
at scala.collection.immutable.Stream$Cons.tail(Stream.scala:1173)
at scala.collection.immutable.Stream$Cons.tail(Stream.scala:1163)
at scala.collection.immutable.Stream.foreach(Stream.scala:534)
at scala.collection.TraversableOnce.count(TraversableOnce.scala:152)
at scala.collection.TraversableOnce.count$(TraversableOnce.scala:145)
at scala.collection.AbstractTraversable.count(Traversable.scala:108)
at
org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection$.createCode(GenerateUnsafeProjection.scala:293)
at
org.apache.spark.sql.execution.aggregate.HashAggregateExec.doConsumeWithKeys(HashAggregateExec.scala:623)
... etc ...
```
When `HashAggregateExec` attempts to bind the references in the group-by
expressions, attribute _groupingexpression#24 is missing from the child
`ProjectExec`'s output.
This is due to the way `PullOutGroupingExpressions`, when determining which
grouping expressions to shift from the `Aggregate` node to a `Project` node,
populates `complexGroupingExpressionMap`. `PullOutGroupingExpressions` uses a
map operation to iterate over `groupingExpressions` and updates
`complexGroupingExpressionMap` in the closure passed to `map()`. However, if
`groupingExpressions` is a `Stream`, the map operation is evaluated lazily, and
isn't fully completed until `Compute [...]
### Does this PR introduce _any_ user-facing change?
No, other than the above query now works.
### How was this patch tested?
New unit test.
Closes #35537 from bersprockets/groupby_stream_issue.
Authored-by: Bruce Robbins <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>
(cherry picked from commit ad2bc7d82296527582dfa469aad33123afdf6736)
Signed-off-by: Hyukjin Kwon <[email protected]>
---
.../spark/sql/catalyst/optimizer/PullOutGroupingExpressions.scala | 2 +-
.../test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala | 5 +++++
2 files changed, 6 insertions(+), 1 deletion(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PullOutGroupingExpressions.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PullOutGroupingExpressions.scala
index 859a73a..1bd186d 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PullOutGroupingExpressions.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PullOutGroupingExpressions.scala
@@ -50,7 +50,7 @@ object PullOutGroupingExpressions extends Rule[LogicalPlan] {
plan.transformWithPruning(_.containsPattern(AGGREGATE)) {
case a: Aggregate if a.resolved =>
val complexGroupingExpressionMap =
mutable.LinkedHashMap.empty[Expression, NamedExpression]
- val newGroupingExpressions = a.groupingExpressions.map {
+ val newGroupingExpressions = a.groupingExpressions.toIndexedSeq.map {
case e if !e.foldable && e.children.nonEmpty =>
complexGroupingExpressionMap
.getOrElseUpdate(e.canonicalized, Alias(e,
s"_groupingexpression")())
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
index 1d12811..c074a20 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
@@ -1427,6 +1427,11 @@ class DataFrameAggregateSuite extends QueryTest
val emptyAgg = Map.empty[String, String]
assert(spark.range(2).where("id > 2").agg(emptyAgg).limit(1).count == 1)
}
+
+ test("SPARK-38221: group by stream of complex expressions should not fail") {
+ val df = Seq(1).toDF("id").groupBy(Stream($"id" + 1, $"id" + 2):
_*).sum("id")
+ checkAnswer(df, Row(2, 3, 1))
+ }
}
case class B(c: Option[Double])
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
