You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2023/01/20 11:17:43 UTC
[GitHub] [spark] EnricoMi opened a new pull request, #39673: [SPARK-42132][SQL] Deduplicate attributes in groupByKey.cogroup
EnricoMi opened a new pull request, #39673:
URL: https://github.com/apache/spark/pull/39673
### What changes were proposed in this pull request?
This deduplicate attributes that exist on both sides of a `CoGroup` by aliasing the occurrence on the right side.
### Why are the changes needed?
Usually, DeduplicateRelations rule does exactly this. But the generic `QueryPlan.rewriteAttrs` replaces all occurrences of the duplicate reference with the new reference, but `CoGroup` uses the old reference for left and right group attributes, value attributes, and group order. Only the occurrences in the right attributes must be replaced.
Further, the right deserialization expression is not touched at all.
The following DataFrame cannot be evaluated:
```scala
val df = spark.range(3)
val left_grouped_df = df.groupBy("id").as[Long, Long]
val right_grouped_df = df.groupBy("id").as[Long, Long]
val cogroup_df = left_grouped_df.cogroup(right_grouped_df) {
case (key, left, right) => left
}
```
The query plan:
```
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- SerializeFromObject [input[0, bigint, false] AS value#12L]
+- CoGroup, id#0: bigint, id#0: bigint, id#0: bigint, [id#13L], [id#13L], [id#13L], [id#13L], obj#11: bigint
:- !Sort [id#13L ASC NULLS FIRST], false, 0
: +- !Exchange hashpartitioning(id#13L, 200), ENSURE_REQUIREMENTS, [plan_id=16]
: +- Range (0, 3, step=1, splits=16)
+- Sort [id#13L ASC NULLS FIRST], false, 0
+- Exchange hashpartitioning(id#13L, 200), ENSURE_REQUIREMENTS, [plan_id=17]
+- Range (0, 3, step=1, splits=16)
```
### Does this PR introduce _any_ user-facing change?
This fixes correctness.
### How was this patch tested?
Unit test in `DataFrameSuite`.
--
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
[GitHub] [spark] AmplabJenkins commented on pull request #39673: [SPARK-42132][SQL] Deduplicate attributes in groupByKey.cogroup
Posted by "AmplabJenkins (via GitHub)" <gi...@apache.org>.
AmplabJenkins commented on PR #39673:
URL: https://github.com/apache/spark/pull/39673#issuecomment-1399633297
Can one of the admins verify this patch?
--
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-42132][SQL] Deduplicate attributes in groupByKey.cogroup [spark]
Posted by "EnricoMi (via GitHub)" <gi...@apache.org>.
EnricoMi commented on PR #39673:
URL: https://github.com/apache/spark/pull/39673#issuecomment-1753026078
Fixed in #41554.
--
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
[GitHub] [spark] shardulm94 commented on pull request #39673: [SPARK-42132][SQL] Deduplicate attributes in groupByKey.cogroup
Posted by "shardulm94 (via GitHub)" <gi...@apache.org>.
shardulm94 commented on PR #39673:
URL: https://github.com/apache/spark/pull/39673#issuecomment-1448587391
cc: @cloud-fan @viirya @gengliangwang For help with reviews.
--
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
[GitHub] [spark] EnricoMi commented on pull request #39673: [SPARK-42132][SQL] Deduplicate attributes in groupByKey.cogroup
Posted by GitBox <gi...@apache.org>.
EnricoMi commented on PR #39673:
URL: https://github.com/apache/spark/pull/39673#issuecomment-1398246138
Ideally, `QueryPlan.rewriteAttrs` would not replace occurrences `id#0L#` with `id#13L` in all fields of `CoGroup`, but only in `rightDeserializer`, `rightGroup`, `rightAttr`, `rightOrder`:
```
case class CoGroup(
func: (Any, Iterator[Any], Iterator[Any]) => TraversableOnce[Any],
keyDeserializer: Expression,
leftDeserializer: Expression,
rightDeserializer: Expression,
leftGroup: Seq[Attribute],
rightGroup: Seq[Attribute],
leftAttr: Seq[Attribute],
rightAttr: Seq[Attribute],
leftOrder: Seq[SortOrder],
rightOrder: Seq[SortOrder],
outputObjAttr: Attribute,
left: LogicalPlan,
right: LogicalPlan) extends BinaryNode with ObjectProducer { ... }
```
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #39673: [SPARK-42132][SQL] Deduplicate attributes in groupByKey.cogroup
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #39673:
URL: https://github.com/apache/spark/pull/39673#discussion_r1177978929
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/object.scala:
##########
@@ -659,22 +659,46 @@ object CoGroup {
right: LogicalPlan): LogicalPlan = {
require(StructType.fromAttributes(leftGroup) == StructType.fromAttributes(rightGroup))
+ // SPARK-42132: The DeduplicateRelations rule would replace duplicate attributes
+ // in the right plan and rewrite rightGroup and rightAttr. But it would also rewrite
+ // leftGroup and leftAttr, which is wrong. Additionally, it does not rewrite rightDeserializer.
+ // Aliasing duplicate attributes in the right plan deduplicates them and stops
+ // DeduplicateRelations to do harm.
+ val duplicateAttributes = AttributeMap(
+ right.output.filter(left.output.contains).map(a => a -> Alias(a, a.name)())
+ )
+
+ def dedup(attrs: Seq[Attribute]): Seq[NamedExpression] =
+ attrs.map(attr => duplicateAttributes.getOrElse(attr, attr))
+
+ // rightOrder is resolved against right plan, so deduplication not needed
+ val (dedupRightGroup, dedupRightAttr, dedupRight) =
+ if (duplicateAttributes.nonEmpty) {
Review Comment:
it's a bit weird to do dedup here. Can we update the `DeduplicateRelations` rule to handle `CoGroup` specially?
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #39673: [SPARK-42132][SQL] Deduplicate attributes in groupByKey.cogroup
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #39673:
URL: https://github.com/apache/spark/pull/39673#discussion_r1278982630
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/object.scala:
##########
@@ -659,22 +659,46 @@ object CoGroup {
right: LogicalPlan): LogicalPlan = {
require(StructType.fromAttributes(leftGroup) == StructType.fromAttributes(rightGroup))
+ // SPARK-42132: The DeduplicateRelations rule would replace duplicate attributes
+ // in the right plan and rewrite rightGroup and rightAttr. But it would also rewrite
+ // leftGroup and leftAttr, which is wrong. Additionally, it does not rewrite rightDeserializer.
+ // Aliasing duplicate attributes in the right plan deduplicates them and stops
+ // DeduplicateRelations to do harm.
+ val duplicateAttributes = AttributeMap(
+ right.output.filter(left.output.contains).map(a => a -> Alias(a, a.name)())
+ )
+
+ def dedup(attrs: Seq[Attribute]): Seq[NamedExpression] =
+ attrs.map(attr => duplicateAttributes.getOrElse(attr, attr))
+
+ // rightOrder is resolved against right plan, so deduplication not needed
+ val (dedupRightGroup, dedupRightAttr, dedupRight) =
+ if (duplicateAttributes.nonEmpty) {
Review Comment:
@Hisoka-X is working on it: https://github.com/apache/spark/pull/41554
--
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
[GitHub] [spark] EnricoMi commented on a diff in pull request #39673: [SPARK-42132][SQL] Deduplicate attributes in groupByKey.cogroup
Posted by "EnricoMi (via GitHub)" <gi...@apache.org>.
EnricoMi commented on code in PR #39673:
URL: https://github.com/apache/spark/pull/39673#discussion_r1178369201
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/object.scala:
##########
@@ -659,22 +659,46 @@ object CoGroup {
right: LogicalPlan): LogicalPlan = {
require(StructType.fromAttributes(leftGroup) == StructType.fromAttributes(rightGroup))
+ // SPARK-42132: The DeduplicateRelations rule would replace duplicate attributes
+ // in the right plan and rewrite rightGroup and rightAttr. But it would also rewrite
+ // leftGroup and leftAttr, which is wrong. Additionally, it does not rewrite rightDeserializer.
+ // Aliasing duplicate attributes in the right plan deduplicates them and stops
+ // DeduplicateRelations to do harm.
+ val duplicateAttributes = AttributeMap(
+ right.output.filter(left.output.contains).map(a => a -> Alias(a, a.name)())
+ )
+
+ def dedup(attrs: Seq[Attribute]): Seq[NamedExpression] =
+ attrs.map(attr => duplicateAttributes.getOrElse(attr, attr))
+
+ // rightOrder is resolved against right plan, so deduplication not needed
+ val (dedupRightGroup, dedupRightAttr, dedupRight) =
+ if (duplicateAttributes.nonEmpty) {
Review Comment:
I don't really see how `DeduplicateRelations` can be modified to not rewrite **all** attributes of CoGroup.
In `DeduplicateRelations.apply` method `renewDuplicatedRelations` is called, which calls `rewriteAttrs(attrMap)` on the `CoGroup`, which rewrites **all** attributes.
If you are suggesting to add `case cogroup @ CoGroup(...) =>` to
https://github.com/apache/spark/blob/f1a6c5ed08a8f787b4924800f2a345b5a5c94dbf/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala#L45-L83
then this won't work because all attributes of `CoGroup` have been rewritten at this point.
I'd appreciate some pointers or sketch of a solution.
--
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-42132][SQL] Deduplicate attributes in groupByKey.cogroup [spark]
Posted by "EnricoMi (via GitHub)" <gi...@apache.org>.
EnricoMi closed pull request #39673: [SPARK-42132][SQL] Deduplicate attributes in groupByKey.cogroup
URL: https://github.com/apache/spark/pull/39673
--
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
[GitHub] [spark] mridulm commented on pull request #39673: [SPARK-42132][SQL] Deduplicate attributes in groupByKey.cogroup
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #39673:
URL: https://github.com/apache/spark/pull/39673#issuecomment-1523111426
+CC @HyukjinKwon, @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
[GitHub] [spark] EnricoMi commented on a diff in pull request #39673: [SPARK-42132][SQL] Deduplicate attributes in groupByKey.cogroup
Posted by "EnricoMi (via GitHub)" <gi...@apache.org>.
EnricoMi commented on code in PR #39673:
URL: https://github.com/apache/spark/pull/39673#discussion_r1178369201
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/object.scala:
##########
@@ -659,22 +659,46 @@ object CoGroup {
right: LogicalPlan): LogicalPlan = {
require(StructType.fromAttributes(leftGroup) == StructType.fromAttributes(rightGroup))
+ // SPARK-42132: The DeduplicateRelations rule would replace duplicate attributes
+ // in the right plan and rewrite rightGroup and rightAttr. But it would also rewrite
+ // leftGroup and leftAttr, which is wrong. Additionally, it does not rewrite rightDeserializer.
+ // Aliasing duplicate attributes in the right plan deduplicates them and stops
+ // DeduplicateRelations to do harm.
+ val duplicateAttributes = AttributeMap(
+ right.output.filter(left.output.contains).map(a => a -> Alias(a, a.name)())
+ )
+
+ def dedup(attrs: Seq[Attribute]): Seq[NamedExpression] =
+ attrs.map(attr => duplicateAttributes.getOrElse(attr, attr))
+
+ // rightOrder is resolved against right plan, so deduplication not needed
+ val (dedupRightGroup, dedupRightAttr, dedupRight) =
+ if (duplicateAttributes.nonEmpty) {
Review Comment:
I don't really see how `DeduplicateRelations` can be modified to not rewrite **all** attributes of CoGroup.
In `DeduplicateRelations.apply` method `renewDuplicatedRelations` is called, which calls `rewriteAttrs(attrMap)` on the `CoGroup`, which rewrites **all** attributes.
If you are suggesting to add `case cogroup @ CoGroup(...) =>` to
https://github.com/apache/spark/blob/f1a6c5ed08a8f787b4924800f2a345b5a5c94dbf/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala#L45-L83
then this won't work because all attributes of `CoGroup` have been rewritten at this point.
--
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