You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "stevomitric (via GitHub)" <gi...@apache.org> on 2024/06/13 12:57:55 UTC
[PR] [SPARK-48619] Fix reference error when grouping by MapType [spark]
stevomitric opened a new pull request, #46974:
URL: https://github.com/apache/spark/pull/46974
### What changes were proposed in this pull request?
In this PR, rule `InsertMapSortInGroupingExpressions` was modified to add an alias for `map_sort(col)` . Grouping and aggregate expressions are then modified to use this alias.
### Why are the changes needed?
Currently, trying to reference a map column that is being grouped by, causes reference errors because we are only replacing the grouping expression `m` with `map_sort(m)`. The following query fails:
```sql
SELECT m FROM table GROUP BY m
```
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
New Tests in this PR, and existing tests regarding grouping of map types.
### 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-48619] Fix reference error when grouping by MapType [spark]
Posted by "n-young-db (via GitHub)" <gi...@apache.org>.
n-young-db commented on code in PR #46974:
URL: https://github.com/apache/spark/pull/46974#discussion_r1644727900
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InsertMapSortInGroupingExpressions.scala:
##########
@@ -17,29 +17,41 @@
package org.apache.spark.sql.catalyst.optimizer
-import org.apache.spark.sql.catalyst.expressions.MapSort
-import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan}
+import org.apache.spark.sql.catalyst.expressions.{Alias, AliasHelper, MapSort}
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan, Project}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.TreePattern.AGGREGATE
import org.apache.spark.sql.types.MapType
/**
- * Adds MapSort to group expressions containing map columns, as the key/value paris need to be
- * in the correct order before grouping:
+ * Adds MapSort to group expressions containing map columns by pushing down the projection,
+ * as the key/value pairs need to be in the correct order before grouping:
* SELECT COUNT(*) FROM TABLE GROUP BY map_column =>
- * SELECT COUNT(*) FROM TABLE GROUP BY map_sort(map_column)
+ * SELECT COUNT(*) FROM (SELECT map_sort(map_column) as map_sorted) GROUP BY map_sorted
+ *
+ * SELECT map_column FROM TABLE GROUP BY map_column =>
+ * SELECT map_sorted FROM (SELECT map_sort(map_column) as map_sorted) GROUP BY map_sorted
+ *
+ * We inject a new `Project` to ensure that we don't do recomputation, and so that we can reference
+ * the newly sorted map column in the aggregate expressions list.
*/
-object InsertMapSortInGroupingExpressions extends Rule[LogicalPlan] {
+object InsertMapSortInGroupingExpressions extends Rule[LogicalPlan] with AliasHelper {
override def apply(plan: LogicalPlan): LogicalPlan = plan.transformWithPruning(
_.containsPattern(AGGREGATE), ruleId) {
- case a @ Aggregate(groupingExpr, _, _) =>
- val newGrouping = groupingExpr.map { expr =>
- if (!expr.isInstanceOf[MapSort] && expr.dataType.isInstanceOf[MapType]) {
- MapSort(expr)
- } else {
- expr
- }
- }
- a.copy(groupingExpressions = newGrouping)
+ case a @ Aggregate(groupingExpr, aggregateExpr, child) =>
+ val toSort = groupingExpr.filter( expr =>
+ !expr.isInstanceOf[MapSort] && expr.dataType.isInstanceOf[MapType]
+ )
+ val newChild = Project(toSort.map( expr =>
Review Comment:
We need to project the other, nonsorted columns, otherwise other attributes will be dropped. Just need to add `child.output` to the Project
```
val toSort = groupingExpr.filter( expr =>
!expr.isInstanceOf[MapSort] && expr.dataType.isInstanceOf[MapType]
)
val newChild = Project(toSort.map( expr =>
Alias(MapSort(expr), s"map_sorted_${expr.prettyName}")()
) ++ child.output, child)
```
and then ColumnPruning will take care of the rest. I missed this, but the tests I suggested above should have caught it
--
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-48619] Fix reference error when grouping by MapType [spark]
Posted by "stevomitric (via GitHub)" <gi...@apache.org>.
stevomitric commented on code in PR #46974:
URL: https://github.com/apache/spark/pull/46974#discussion_r1644407740
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InsertMapSortInGroupingExpressions.scala:
##########
@@ -17,29 +17,41 @@
package org.apache.spark.sql.catalyst.optimizer
-import org.apache.spark.sql.catalyst.expressions.MapSort
-import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan}
+import org.apache.spark.sql.catalyst.expressions.{Alias, AliasHelper, MapSort}
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan, Project}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.TreePattern.AGGREGATE
import org.apache.spark.sql.types.MapType
/**
- * Adds MapSort to group expressions containing map columns, as the key/value paris need to be
- * in the correct order before grouping:
+ * Adds MapSort to group expressions containing map columns by pushing down the projection,
+ * as the key/value pairs need to be in the correct order before grouping:
* SELECT COUNT(*) FROM TABLE GROUP BY map_column =>
- * SELECT COUNT(*) FROM TABLE GROUP BY map_sort(map_column)
+ * SELECT COUNT(*) FROM (SELECT map_sort(map_column) as map_sorted) GROUP BY map_sorted
+ *
+ * SELECT map_column FROM TABLE GROUP BY map_column =>
+ * SELECT map_sorted FROM (SELECT map_sort(map_column) as map_sorted) GROUP BY map_sorted
+ *
+ * We inject a new `Project` to ensure that we don't do recomputation, and so that we can reference
+ * the newly sorted map column in the aggregate expressions list.
*/
-object InsertMapSortInGroupingExpressions extends Rule[LogicalPlan] {
+object InsertMapSortInGroupingExpressions extends Rule[LogicalPlan] with AliasHelper {
override def apply(plan: LogicalPlan): LogicalPlan = plan.transformWithPruning(
_.containsPattern(AGGREGATE), ruleId) {
- case a @ Aggregate(groupingExpr, _, _) =>
- val newGrouping = groupingExpr.map { expr =>
- if (!expr.isInstanceOf[MapSort] && expr.dataType.isInstanceOf[MapType]) {
- MapSort(expr)
- } else {
- expr
- }
- }
- a.copy(groupingExpressions = newGrouping)
+ case a @ Aggregate(groupingExpr, aggregateExpr, child) =>
+ val toSort = groupingExpr.filter( expr =>
+ !expr.isInstanceOf[MapSort] && expr.dataType.isInstanceOf[MapType]
+ )
+ val newChild = Project(toSort.map( expr =>
+ Alias(MapSort(expr), "map_sorted")()
Review Comment:
Inserted `expr.prettyName` here.
--
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-48619] Fix reference error when grouping by MapType [spark]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #46974: [SPARK-48619] Fix reference error when grouping by MapType
URL: https://github.com/apache/spark/pull/46974
--
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-48619] Fix reference error when grouping by MapType [spark]
Posted by "n-young-db (via GitHub)" <gi...@apache.org>.
n-young-db commented on code in PR #46974:
URL: https://github.com/apache/spark/pull/46974#discussion_r1644727900
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InsertMapSortInGroupingExpressions.scala:
##########
@@ -17,29 +17,41 @@
package org.apache.spark.sql.catalyst.optimizer
-import org.apache.spark.sql.catalyst.expressions.MapSort
-import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan}
+import org.apache.spark.sql.catalyst.expressions.{Alias, AliasHelper, MapSort}
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan, Project}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.TreePattern.AGGREGATE
import org.apache.spark.sql.types.MapType
/**
- * Adds MapSort to group expressions containing map columns, as the key/value paris need to be
- * in the correct order before grouping:
+ * Adds MapSort to group expressions containing map columns by pushing down the projection,
+ * as the key/value pairs need to be in the correct order before grouping:
* SELECT COUNT(*) FROM TABLE GROUP BY map_column =>
- * SELECT COUNT(*) FROM TABLE GROUP BY map_sort(map_column)
+ * SELECT COUNT(*) FROM (SELECT map_sort(map_column) as map_sorted) GROUP BY map_sorted
+ *
+ * SELECT map_column FROM TABLE GROUP BY map_column =>
+ * SELECT map_sorted FROM (SELECT map_sort(map_column) as map_sorted) GROUP BY map_sorted
+ *
+ * We inject a new `Project` to ensure that we don't do recomputation, and so that we can reference
+ * the newly sorted map column in the aggregate expressions list.
*/
-object InsertMapSortInGroupingExpressions extends Rule[LogicalPlan] {
+object InsertMapSortInGroupingExpressions extends Rule[LogicalPlan] with AliasHelper {
override def apply(plan: LogicalPlan): LogicalPlan = plan.transformWithPruning(
_.containsPattern(AGGREGATE), ruleId) {
- case a @ Aggregate(groupingExpr, _, _) =>
- val newGrouping = groupingExpr.map { expr =>
- if (!expr.isInstanceOf[MapSort] && expr.dataType.isInstanceOf[MapType]) {
- MapSort(expr)
- } else {
- expr
- }
- }
- a.copy(groupingExpressions = newGrouping)
+ case a @ Aggregate(groupingExpr, aggregateExpr, child) =>
+ val toSort = groupingExpr.filter( expr =>
+ !expr.isInstanceOf[MapSort] && expr.dataType.isInstanceOf[MapType]
+ )
+ val newChild = Project(toSort.map( expr =>
Review Comment:
We need to project the other, nonsorted columns, otherwise other attributes will be dropped. Just need to add `child.output` to the Project
```
val toSort = groupingExpr.filter( expr =>
!expr.isInstanceOf[MapSort] && expr.dataType.isInstanceOf[MapType]
)
val newChild = Project(toSort.map( expr =>
Alias(MapSort(expr), s"map_sorted_${expr.prettyName}")()
) ++ child.output, child)
```
and then ColumnPruning will take care of the rest
--
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-48619] Fix reference error when grouping by MapType [spark]
Posted by "n-young-db (via GitHub)" <gi...@apache.org>.
n-young-db commented on code in PR #46974:
URL: https://github.com/apache/spark/pull/46974#discussion_r1638310150
##########
sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala:
##########
@@ -2202,6 +2202,36 @@ class DataFrameAggregateSuite extends QueryTest
assertAggregateOnDataframe(dfDifferent, numRows, "m0")
}
+ test("Fix reference error when grouping by MapType") {
Review Comment:
Let's have some tests for
- multiple group-by columns
- multiple maptype group-by columns
- maptype in aggregate expressions
- maptype in non-aggregate expression, but in another expression
- selecting other rows
to confirm behavior and non breaking
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InsertMapSortInGroupingExpressions.scala:
##########
@@ -17,29 +17,41 @@
package org.apache.spark.sql.catalyst.optimizer
-import org.apache.spark.sql.catalyst.expressions.MapSort
-import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan}
+import org.apache.spark.sql.catalyst.expressions.{Alias, AliasHelper, MapSort}
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan, Project}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.TreePattern.AGGREGATE
import org.apache.spark.sql.types.MapType
/**
- * Adds MapSort to group expressions containing map columns, as the key/value paris need to be
- * in the correct order before grouping:
+ * Adds MapSort to group expressions containing map columns by pushing down the projection,
+ * as the key/value pairs need to be in the correct order before grouping:
* SELECT COUNT(*) FROM TABLE GROUP BY map_column =>
- * SELECT COUNT(*) FROM TABLE GROUP BY map_sort(map_column)
+ * SELECT COUNT(*) FROM (SELECT map_sort(map_column) as map_sorted) GROUP BY map_sorted
+ *
+ * SELECT map_column FROM TABLE GROUP BY map_column =>
+ * SELECT map_sorted FROM (SELECT map_sort(map_column) as map_sorted) GROUP BY map_sorted
+ *
+ * We inject a new `Project` to ensure that we don't do recomputation, and so that we can reference
+ * the newly sorted map column in the aggregate expressions list.
*/
-object InsertMapSortInGroupingExpressions extends Rule[LogicalPlan] {
+object InsertMapSortInGroupingExpressions extends Rule[LogicalPlan] with AliasHelper {
override def apply(plan: LogicalPlan): LogicalPlan = plan.transformWithPruning(
_.containsPattern(AGGREGATE), ruleId) {
- case a @ Aggregate(groupingExpr, _, _) =>
- val newGrouping = groupingExpr.map { expr =>
- if (!expr.isInstanceOf[MapSort] && expr.dataType.isInstanceOf[MapType]) {
- MapSort(expr)
- } else {
- expr
- }
- }
- a.copy(groupingExpressions = newGrouping)
+ case a @ Aggregate(groupingExpr, aggregateExpr, child) =>
+ val toSort = groupingExpr.filter( expr =>
+ !expr.isInstanceOf[MapSort] && expr.dataType.isInstanceOf[MapType]
+ )
+ val newChild = Project(toSort.map( expr =>
+ Alias(MapSort(expr), "map_sorted")()
Review Comment:
Might consider adding the name of `expr` into the alias so it makes it easier to track up the plan
--
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-48619] Fix reference error when grouping by MapType [spark]
Posted by "n-young-db (via GitHub)" <gi...@apache.org>.
n-young-db commented on code in PR #46974:
URL: https://github.com/apache/spark/pull/46974#discussion_r1644727900
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InsertMapSortInGroupingExpressions.scala:
##########
@@ -17,29 +17,41 @@
package org.apache.spark.sql.catalyst.optimizer
-import org.apache.spark.sql.catalyst.expressions.MapSort
-import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan}
+import org.apache.spark.sql.catalyst.expressions.{Alias, AliasHelper, MapSort}
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan, Project}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.TreePattern.AGGREGATE
import org.apache.spark.sql.types.MapType
/**
- * Adds MapSort to group expressions containing map columns, as the key/value paris need to be
- * in the correct order before grouping:
+ * Adds MapSort to group expressions containing map columns by pushing down the projection,
+ * as the key/value pairs need to be in the correct order before grouping:
* SELECT COUNT(*) FROM TABLE GROUP BY map_column =>
- * SELECT COUNT(*) FROM TABLE GROUP BY map_sort(map_column)
+ * SELECT COUNT(*) FROM (SELECT map_sort(map_column) as map_sorted) GROUP BY map_sorted
+ *
+ * SELECT map_column FROM TABLE GROUP BY map_column =>
+ * SELECT map_sorted FROM (SELECT map_sort(map_column) as map_sorted) GROUP BY map_sorted
+ *
+ * We inject a new `Project` to ensure that we don't do recomputation, and so that we can reference
+ * the newly sorted map column in the aggregate expressions list.
*/
-object InsertMapSortInGroupingExpressions extends Rule[LogicalPlan] {
+object InsertMapSortInGroupingExpressions extends Rule[LogicalPlan] with AliasHelper {
override def apply(plan: LogicalPlan): LogicalPlan = plan.transformWithPruning(
_.containsPattern(AGGREGATE), ruleId) {
- case a @ Aggregate(groupingExpr, _, _) =>
- val newGrouping = groupingExpr.map { expr =>
- if (!expr.isInstanceOf[MapSort] && expr.dataType.isInstanceOf[MapType]) {
- MapSort(expr)
- } else {
- expr
- }
- }
- a.copy(groupingExpressions = newGrouping)
+ case a @ Aggregate(groupingExpr, aggregateExpr, child) =>
+ val toSort = groupingExpr.filter( expr =>
+ !expr.isInstanceOf[MapSort] && expr.dataType.isInstanceOf[MapType]
+ )
+ val newChild = Project(toSort.map( expr =>
Review Comment:
We need to project the other, nonsorted columns. Probably something like the following is best:
```
val (toSort, rest) = groupingExpr.partition( expr =>
!expr.isInstanceOf[MapSort] && expr.dataType.isInstanceOf[MapType]
)
val newChild = Project(toSort.map( expr =>
Alias(MapSort(expr), s"map_sorted_${expr.prettyName}")()
) ++ rest, child)
```
--
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-48619] Fix reference error when grouping by MapType [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #46974:
URL: https://github.com/apache/spark/pull/46974#discussion_r1642256130
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InsertMapSortInGroupingExpressions.scala:
##########
@@ -17,29 +17,41 @@
package org.apache.spark.sql.catalyst.optimizer
-import org.apache.spark.sql.catalyst.expressions.MapSort
-import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan}
+import org.apache.spark.sql.catalyst.expressions.{Alias, AliasHelper, MapSort}
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan, Project}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.TreePattern.AGGREGATE
import org.apache.spark.sql.types.MapType
/**
- * Adds MapSort to group expressions containing map columns, as the key/value paris need to be
- * in the correct order before grouping:
+ * Adds MapSort to group expressions containing map columns by pushing down the projection,
+ * as the key/value pairs need to be in the correct order before grouping:
* SELECT COUNT(*) FROM TABLE GROUP BY map_column =>
- * SELECT COUNT(*) FROM TABLE GROUP BY map_sort(map_column)
+ * SELECT COUNT(*) FROM (SELECT map_sort(map_column) as map_sorted) GROUP BY map_sorted
+ *
+ * SELECT map_column FROM TABLE GROUP BY map_column =>
+ * SELECT map_sorted FROM (SELECT map_sort(map_column) as map_sorted) GROUP BY map_sorted
+ *
+ * We inject a new `Project` to ensure that we don't do recomputation, and so that we can reference
+ * the newly sorted map column in the aggregate expressions list.
*/
-object InsertMapSortInGroupingExpressions extends Rule[LogicalPlan] {
+object InsertMapSortInGroupingExpressions extends Rule[LogicalPlan] with AliasHelper {
override def apply(plan: LogicalPlan): LogicalPlan = plan.transformWithPruning(
_.containsPattern(AGGREGATE), ruleId) {
- case a @ Aggregate(groupingExpr, _, _) =>
- val newGrouping = groupingExpr.map { expr =>
- if (!expr.isInstanceOf[MapSort] && expr.dataType.isInstanceOf[MapType]) {
- MapSort(expr)
- } else {
- expr
- }
- }
- a.copy(groupingExpressions = newGrouping)
+ case a @ Aggregate(groupingExpr, aggregateExpr, child) =>
+ val toSort = groupingExpr.filter( expr =>
+ !expr.isInstanceOf[MapSort] && expr.dataType.isInstanceOf[MapType]
+ )
+ val newChild = Project(toSort.map( expr =>
Review Comment:
shouldn't we add `child.output` in the project list as well? How can we make sure the `Aggregate` only references columns of `toSort`?
--
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-48619] Fix reference error when grouping by MapType [spark]
Posted by "stevomitric (via GitHub)" <gi...@apache.org>.
stevomitric commented on code in PR #46974:
URL: https://github.com/apache/spark/pull/46974#discussion_r1644407166
##########
sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala:
##########
@@ -2202,6 +2202,36 @@ class DataFrameAggregateSuite extends QueryTest
assertAggregateOnDataframe(dfDifferent, numRows, "m0")
}
+ test("Fix reference error when grouping by MapType") {
Review Comment:
Added a couple of more tests.
--
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-48619] Fix reference error when grouping by MapType [spark]
Posted by "n-young-db (via GitHub)" <gi...@apache.org>.
n-young-db commented on PR #46974:
URL: https://github.com/apache/spark/pull/46974#issuecomment-2168353399
Approved but please address comments! Just didn't want to block over the weekend.
--
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-48619] Fix reference error when grouping by MapType [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #46974:
URL: https://github.com/apache/spark/pull/46974#discussion_r1645297165
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InsertMapSortInGroupingExpressions.scala:
##########
@@ -17,29 +17,41 @@
package org.apache.spark.sql.catalyst.optimizer
-import org.apache.spark.sql.catalyst.expressions.MapSort
-import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan}
+import org.apache.spark.sql.catalyst.expressions.{Alias, AliasHelper, MapSort}
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan, Project}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.TreePattern.AGGREGATE
import org.apache.spark.sql.types.MapType
/**
- * Adds MapSort to group expressions containing map columns, as the key/value paris need to be
- * in the correct order before grouping:
+ * Adds MapSort to group expressions containing map columns by pushing down the projection,
+ * as the key/value pairs need to be in the correct order before grouping:
* SELECT COUNT(*) FROM TABLE GROUP BY map_column =>
- * SELECT COUNT(*) FROM TABLE GROUP BY map_sort(map_column)
+ * SELECT COUNT(*) FROM (SELECT map_sort(map_column) as map_sorted) GROUP BY map_sorted
+ *
+ * SELECT map_column FROM TABLE GROUP BY map_column =>
+ * SELECT map_sorted FROM (SELECT map_sort(map_column) as map_sorted) GROUP BY map_sorted
Review Comment:
The problem with this rewrite is it changes the final output schema (column name changed from `map_column` to `map_sorted`). We should alias `map_sort(map_column)` to the original name of the grouping column.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InsertMapSortInGroupingExpressions.scala:
##########
@@ -17,29 +17,41 @@
package org.apache.spark.sql.catalyst.optimizer
-import org.apache.spark.sql.catalyst.expressions.MapSort
-import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan}
+import org.apache.spark.sql.catalyst.expressions.{Alias, AliasHelper, MapSort}
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan, Project}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.TreePattern.AGGREGATE
import org.apache.spark.sql.types.MapType
/**
- * Adds MapSort to group expressions containing map columns, as the key/value paris need to be
- * in the correct order before grouping:
+ * Adds MapSort to group expressions containing map columns by pushing down the projection,
+ * as the key/value pairs need to be in the correct order before grouping:
* SELECT COUNT(*) FROM TABLE GROUP BY map_column =>
- * SELECT COUNT(*) FROM TABLE GROUP BY map_sort(map_column)
+ * SELECT COUNT(*) FROM (SELECT map_sort(map_column) as map_sorted) GROUP BY map_sorted
+ *
+ * SELECT map_column FROM TABLE GROUP BY map_column =>
+ * SELECT map_sorted FROM (SELECT map_sort(map_column) as map_sorted) GROUP BY map_sorted
Review Comment:
The problem with this rewrite is it changes the final output schema (column name changed from `map_column` to `map_sorted`). We should alias `map_sort(map_column)` to the original name of the grouping expression if it's a subclass of `NamedExpression`.
--
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-48619] Fix reference error when grouping by MapType [spark]
Posted by "stevomitric (via GitHub)" <gi...@apache.org>.
stevomitric commented on PR #46974:
URL: https://github.com/apache/spark/pull/46974#issuecomment-2165597478
@n-young-db @cloud-fan can you please take a look at this.
--
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-48619] Fix reference error when grouping by MapType [spark]
Posted by "stevomitric (via GitHub)" <gi...@apache.org>.
stevomitric commented on code in PR #46974:
URL: https://github.com/apache/spark/pull/46974#discussion_r1644411218
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InsertMapSortInGroupingExpressions.scala:
##########
@@ -17,29 +17,41 @@
package org.apache.spark.sql.catalyst.optimizer
-import org.apache.spark.sql.catalyst.expressions.MapSort
-import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan}
+import org.apache.spark.sql.catalyst.expressions.{Alias, AliasHelper, MapSort}
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan, Project}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.TreePattern.AGGREGATE
import org.apache.spark.sql.types.MapType
/**
- * Adds MapSort to group expressions containing map columns, as the key/value paris need to be
- * in the correct order before grouping:
+ * Adds MapSort to group expressions containing map columns by pushing down the projection,
+ * as the key/value pairs need to be in the correct order before grouping:
* SELECT COUNT(*) FROM TABLE GROUP BY map_column =>
- * SELECT COUNT(*) FROM TABLE GROUP BY map_sort(map_column)
+ * SELECT COUNT(*) FROM (SELECT map_sort(map_column) as map_sorted) GROUP BY map_sorted
+ *
+ * SELECT map_column FROM TABLE GROUP BY map_column =>
+ * SELECT map_sorted FROM (SELECT map_sort(map_column) as map_sorted) GROUP BY map_sorted
+ *
+ * We inject a new `Project` to ensure that we don't do recomputation, and so that we can reference
+ * the newly sorted map column in the aggregate expressions list.
*/
-object InsertMapSortInGroupingExpressions extends Rule[LogicalPlan] {
+object InsertMapSortInGroupingExpressions extends Rule[LogicalPlan] with AliasHelper {
override def apply(plan: LogicalPlan): LogicalPlan = plan.transformWithPruning(
_.containsPattern(AGGREGATE), ruleId) {
- case a @ Aggregate(groupingExpr, _, _) =>
- val newGrouping = groupingExpr.map { expr =>
- if (!expr.isInstanceOf[MapSort] && expr.dataType.isInstanceOf[MapType]) {
- MapSort(expr)
- } else {
- expr
- }
- }
- a.copy(groupingExpressions = newGrouping)
+ case a @ Aggregate(groupingExpr, aggregateExpr, child) =>
+ val toSort = groupingExpr.filter( expr =>
+ !expr.isInstanceOf[MapSort] && expr.dataType.isInstanceOf[MapType]
+ )
+ val newChild = Project(toSort.map( expr =>
Review Comment:
Can you elaborate a bit? For aggregate, `toSort` expressions are only taken into consideration - others are left as-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-48619] Fix reference error when grouping by MapType [spark]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #46974:
URL: https://github.com/apache/spark/pull/46974#issuecomment-2380313073
We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!
--
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