Skip to content

[SPARK-43781][SQL] Fix IllegalStateException when cogrouping two datasets derived from the same source #41554

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom
Closed
wants to merge 14 commits into from
Closed

[SPARK-43781][SQL] Fix IllegalStateException when cogrouping two datasets derived from the same source #41554

wants to merge 14 commits into from

Conversation

@Hisoka-X
Copy link
Member

@Hisoka-X Hisoka-X commented Jun 12, 2023 •
edited
Loading

What changes were proposed in this pull request?

When cogroup two datasets derived from same source, eg:

val inputType = StructType(Array(StructField("id", LongType, false),
   StructField("type", StringType, false)))
val keyType = StructType(Array(StructField("id", LongType, false)))

val inputRows = new java.util.ArrayList[Row]()
inputRows.add(Row(1L, "foo"))
inputRows.add(Row(1L, "bar"))
inputRows.add(Row(2L, "foo"))
val input = spark.createDataFrame(inputRows, inputType)
val fooGroups = input.filter("type = 'foo'").groupBy("id").as(RowEncoder(keyType),
   RowEncoder(inputType))
val barGroups = input.filter("type = 'bar'").groupBy("id").as(RowEncoder(keyType),
   RowEncoder(inputType))

val result = fooGroups.cogroup(barGroups) { case (row, iterator, iterator1) =>
   iterator.toSeq ++ iterator1.toSeq
}(RowEncoder(inputType)).collect()

The error will be reported:

21:03:27.651 ERROR org.apache.spark.executor.Executor: Exception in task 1.0 in stage 0.0 (TID 1)
java.lang.IllegalStateException: Couldn't find id#19L in [id#0L,type#1]
	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)
	...

The reason are DeduplicateRelations rewrite LocalRelation but can't rewrite left(right)Group and left(right)Attr in CoGroup. In fact, the Join will face same situation. But Join regenerate plan when invoke itself to avoid this situation. Please refer https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala#L1089

This PR let DeduplicateRelations handle with CoGroup case

Why are the changes needed?

Fix IllegalStateException when cogrouping two datasets derived from the same source

Does this PR introduce any user-facing change?

No

How was this patch tested?

Add new test

@github-actions github-actions bot added the SQL label Jun 12, 2023
@Hisoka-X
Copy link
Member Author

Hisoka-X commented Jun 12, 2023 •
edited
Loading

The logical plan before this PR. The left(right)Group and left(right)Attr in CoGroup not be rewrited.

== Analyzed Logical Plan ==
id: bigint, type: string
SerializeFromObject [assertnotnull(validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 0, id), LongType, ObjectType(class java.lang.Long)).longValue) AS id#17L, assertnotnull(staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 1, type), StringType, ObjectType(class java.lang.String)), true, false, true)) AS type#18]
+- CoGroup org.apache.spark.sql.DatasetSuite$$Lambda$1658/631349266@5aa62ee7, createexternalrow(staticinvoke(class java.lang.Long, ObjectType(class java.lang.Long), valueOf, id#0L, true, false, true), StructField(id,LongType,false)), createexternalrow(staticinvoke(class java.lang.Long, ObjectType(class java.lang.Long), valueOf, id#0L, true, false, true), type#1.toString, StructField(id,LongType,false), StructField(type,StringType,false)), createexternalrow(staticinvoke(class java.lang.Long, ObjectType(class java.lang.Long), valueOf, id#0L, true, false, true), type#1.toString, StructField(id,LongType,false), StructField(type,StringType,false)), [id#19L], [id#19L], [id#19L, type#20], [id#19L, type#20], obj#16: org.apache.spark.sql.Row
   :- Project [id#0L, type#1]
   :  +- Filter (type#1 = foo)
   :     +- LocalRelation [id#0L, type#1]
   +- Project [id#19L, type#20]
      +- Filter (type#20 = bar)
         +- LocalRelation [id#19L, type#20]

@Hisoka-X Hisoka-X force-pushed the SPARK-43781_cogrouping_two_datasets branch 2 times, most recently from 00b8dbf to 0d7cd2f Compare June 13, 2023 04:34
@Hisoka-X
Copy link
Member Author

Hisoka-X commented Jun 13, 2023

cc @cloud-fan @dongjoon-hyun

@cloud-fan
Copy link
Contributor

cloud-fan commented Jun 19, 2023

can we make DeduplicateRelations handle CoGroup?

@Hisoka-X
Copy link
Member Author

Hisoka-X commented Jun 19, 2023

can we make DeduplicateRelations handle CoGroup?

Ok for me, before do this, please check #41347

@Hisoka-X Hisoka-X force-pushed the SPARK-43781_cogrouping_two_datasets branch from 0d7cd2f to 83c1382 Compare July 11, 2023 04:10
@Hisoka-X
Copy link
Member Author

Hisoka-X commented Jul 11, 2023

Hi @cloud-fan , I let DeduplicateRelations handle CoGroup at now. Please recheck again. Thanks!

@Hisoka-X Hisoka-X force-pushed the SPARK-43781_cogrouping_two_datasets branch from 2ad8493 to d45eab2 Compare July 20, 2023 02:26
@cloud-fan cloud-fan mentioned this pull request Jul 31, 2023
Closed
@Hisoka-X
Copy link
Member Author

Hisoka-X commented Aug 7, 2023

kindly ping @cloud-fan

cloud-fan
cloud-fan reviewed Aug 8, 2023
View reviewed changes
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala Outdated
case c: CoGroup =>
// SPARK-43781: CoGroup is a special case, as it has different output attributes
// from its children. We need to update the output attributes of CoGroup manually.
val newLeftAttr = c.leftAttr.map(attr => attrMap.getOrElse(attr, attr))
Copy link
Contributor

@cloud-fan cloud-fan Aug 8, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why the left-right attributes does not need to call rewriteAttrsMatchWithSubPlan?

Copy link
Member Author

@Hisoka-X Hisoka-X Aug 8, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just changed by review #41554 (comment)

Copy link
Contributor

@cloud-fan cloud-fan Aug 9, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should the same apple to newRightAttr?

Copy link
Contributor

@cloud-fan cloud-fan Aug 9, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

and to left/right group?

Copy link
Member Author

@Hisoka-X Hisoka-X Aug 9, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should the same apple to newRightAttr?

In fact it will be more safer, but I can't produce any negetive case.

and to left/right group?

what's meaning? Sorry I don't get it. The left/right group already apply rewriteAttrsMatchWithSubPlan.

Copy link
Member Author

@Hisoka-X Hisoka-X Aug 9, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

oh, if you mean the leftAttr should same to rightAttr and left/right group. I would say yes. Maybe we should add rewriteAttrsMatchWithSubPlan to newLeftAttr. After code refactoring, invoke rewriteAttrsMatchWithSubPlan not a big deal. So I add it back.

HyukjinKwon
HyukjinKwon reviewed Aug 9, 2023
View reviewed changes
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala Outdated

planWithNewChildren match {
case c: CoGroup =>
// SPARK-43781: CoGroup is a special case, as it has different output attributes
Copy link
Member

@HyukjinKwon HyukjinKwon Aug 9, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What do you mean by "it has different output attributes"?

There are similar cases where it creates new output attributes (not from the child): MapPartitionsInR, MapPartitionsInRWithArrow, MapElements, MapGroups, FlatMapGroupsWithState, FlatMapGroupsInR, FlatMapGroupsInR, FlatMapGroupsInRWithArrow, FlatMapGroupsInPandas, MapInPandas, PythonMapInArrow, and FlatMapGroupsInPandasWithState.

There's another CoGroup operation in Python (cogroup.apply() that is FlatMapCoGroupsInPandas). Would be great to double check.

Copy link
Member Author

@Hisoka-X Hisoka-X Aug 9, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What do you mean by "it has different output attributes"?

The describe not clear, let me update it.

There are similar cases where it creates new output attributes (not from the child): MapPartitionsInR, MapPartitionsInRWithArrow, MapElements, MapGroups, FlatMapGroupsWithState, FlatMapGroupsInR, FlatMapGroupsInR, FlatMapGroupsInRWithArrow, FlatMapGroupsInPandas, MapInPandas, PythonMapInArrow, and FlatMapGroupsInPandasWithState.
There's another CoGroup operation in Python (cogroup.apply() that is FlatMapCoGroupsInPandas). Would be great to double check.

Can I do it on another PR? Seem like produce negative case will use some time.

Copy link
Member

@HyukjinKwon HyukjinKwon Aug 10, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yup, that's fine to do them as a followup.

@cloud-fan
Copy link
Contributor

cloud-fan commented Aug 10, 2023 •
edited
Loading

thanks, merging to master!

@cloud-fan cloud-fan closed this in 5db8778 Aug 10, 2023
@Hisoka-X
Copy link
Member Author

Hisoka-X commented Aug 10, 2023

Thanks @cloud-fan @HyukjinKwon

@Hisoka-X Hisoka-X deleted the SPARK-43781_cogrouping_two_datasets branch August 11, 2023 00:48
@Hisoka-X
Copy link
Member Author

Hisoka-X commented Aug 11, 2023

Hi @cloud-fan , seem like the jira ticket not update. https://issues.apache.org/jira/browse/SPARK-43781

hvanhovell pushed a commit to hvanhovell/spark that referenced this pull request Aug 13, 2023
@Hisoka-X @hvanhovell
[SPARK-43781][SQL] Fix IllegalStateException when cogrouping two data…
56a52af 
…sets derived from the same source

### What changes were proposed in this pull request?
When cogroup two datasets derived from same source, eg:
```scala
val inputType = StructType(Array(StructField("id", LongType, false),
   StructField("type", StringType, false)))
val keyType = StructType(Array(StructField("id", LongType, false)))

val inputRows = new java.util.ArrayList[Row]()
inputRows.add(Row(1L, "foo"))
inputRows.add(Row(1L, "bar"))
inputRows.add(Row(2L, "foo"))
val input = spark.createDataFrame(inputRows, inputType)
val fooGroups = input.filter("type = 'foo'").groupBy("id").as(RowEncoder(keyType),
   RowEncoder(inputType))
val barGroups = input.filter("type = 'bar'").groupBy("id").as(RowEncoder(keyType),
   RowEncoder(inputType))

val result = fooGroups.cogroup(barGroups) { case (row, iterator, iterator1) =>
   iterator.toSeq ++ iterator1.toSeq
}(RowEncoder(inputType)).collect()
```
The error will be reported:
```
21:03:27.651 ERROR org.apache.spark.executor.Executor: Exception in task 1.0 in stage 0.0 (TID 1)
java.lang.IllegalStateException: Couldn't find id#19L in [id#0L,type#1]
	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)
	...
```
The reason are `DeduplicateRelations` rewrite `LocalRelation` but can't rewrite `left(right)Group` and `left(right)Attr` in `CoGroup`. In fact, the `Join` will face same situation. But `Join` regenerate plan when invoke itself to avoid this situation. Please refer https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala#L1089

This PR let `DeduplicateRelations` handle with `CoGroup` case

### Why are the changes needed?
Fix IllegalStateException when cogrouping two datasets derived from the same source

### Does this PR introduce _any_ user-facing change?
No

### How was this patch tested?
Add new test

Closes apache#41554 from Hisoka-X/SPARK-43781_cogrouping_two_datasets.

Authored-by: Jia Fan <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
valentinp17 pushed a commit to valentinp17/spark that referenced this pull request Aug 24, 2023
@Hisoka-X
[SPARK-43781][SQL] Fix IllegalStateException when cogrouping two data…
211296b 
…sets derived from the same source

### What changes were proposed in this pull request?
When cogroup two datasets derived from same source, eg:
```scala
val inputType = StructType(Array(StructField("id", LongType, false),
   StructField("type", StringType, false)))
val keyType = StructType(Array(StructField("id", LongType, false)))

val inputRows = new java.util.ArrayList[Row]()
inputRows.add(Row(1L, "foo"))
inputRows.add(Row(1L, "bar"))
inputRows.add(Row(2L, "foo"))
val input = spark.createDataFrame(inputRows, inputType)
val fooGroups = input.filter("type = 'foo'").groupBy("id").as(RowEncoder(keyType),
   RowEncoder(inputType))
val barGroups = input.filter("type = 'bar'").groupBy("id").as(RowEncoder(keyType),
   RowEncoder(inputType))

val result = fooGroups.cogroup(barGroups) { case (row, iterator, iterator1) =>
   iterator.toSeq ++ iterator1.toSeq
}(RowEncoder(inputType)).collect()
```
The error will be reported:
```
21:03:27.651 ERROR org.apache.spark.executor.Executor: Exception in task 1.0 in stage 0.0 (TID 1)
java.lang.IllegalStateException: Couldn't find id#19L in [id#0L,type#1]
	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)
	...
```
The reason are `DeduplicateRelations` rewrite `LocalRelation` but can't rewrite `left(right)Group` and `left(right)Attr` in `CoGroup`. In fact, the `Join` will face same situation. But `Join` regenerate plan when invoke itself to avoid this situation. Please refer https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala#L1089

This PR let `DeduplicateRelations` handle with `CoGroup` case

### Why are the changes needed?
Fix IllegalStateException when cogrouping two datasets derived from the same source

### Does this PR introduce _any_ user-facing change?
No

### How was this patch tested?
Add new test

Closes apache#41554 from Hisoka-X/SPARK-43781_cogrouping_two_datasets.

Authored-by: Jia Fan <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@cloud-fan cloud-fan cloud-fan approved these changes

@HyukjinKwon HyukjinKwon HyukjinKwon left review comments

Assignees

No one assigned

Labels

SQL

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

3 participants

@Hisoka-X @cloud-fan @HyukjinKwon