[SPARK-33822][SQL] Use the `CastSupport.cast` method in HashJoin #30818

maropu · 2020-12-17T07:00:57Z

What changes were proposed in this pull request?

This PR intends to fix the bug that throws a unsupported exception when running the TPCDS q5 with AQE enabled (this option is enabled by default now via SPARK-33679):

java.lang.UnsupportedOperationException: BroadcastExchange does not support the execute() code path.
  at org.apache.spark.sql.execution.exchange.BroadcastExchangeExec.doExecute(BroadcastExchangeExec.scala:189)
  at org.apache.spark.sql.execution.SparkPlan.$anonfun$execute$1(SparkPlan.scala:180)
  at org.apache.spark.sql.execution.SparkPlan.$anonfun$executeQuery$1(SparkPlan.scala:218)
  at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
  at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:215)
  at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:176)
  at org.apache.spark.sql.execution.exchange.ReusedExchangeExec.doExecute(Exchange.scala:60)
  at org.apache.spark.sql.execution.SparkPlan.$anonfun$execute$1(SparkPlan.scala:180)
  at org.apache.spark.sql.execution.SparkPlan.$anonfun$executeQuery$1(SparkPlan.scala:218)
  at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
  at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:215)
  at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:176)
  at org.apache.spark.sql.execution.adaptive.QueryStageExec.doExecute(QueryStageExec.scala:115)
  at org.apache.spark.sql.execution.SparkPlan.$anonfun$execute$1(SparkPlan.scala:180)
  at org.apache.spark.sql.execution.SparkPlan.$anonfun$executeQuery$1(SparkPlan.scala:218)
  at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
  at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:215)
  at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:176)
  at org.apache.spark.sql.execution.SparkPlan.getByteArrayRdd(SparkPlan.scala:321)
  at org.apache.spark.sql.execution.SparkPlan.executeCollectIterator(SparkPlan.scala:397)
  at org.apache.spark.sql.execution.exchange.BroadcastExchangeExec.$anonfun$relationFuture$1(BroadcastExchangeExec.scala:118)
  at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withThreadLocalCaptured$1(SQLExecution.scala:185)
  at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
  ...

I've checked the AQE code and I found EnsureRequirements wrongly puts BroadcastExchange on a top of BroadcastQueryStage in the reOptimize phase as follows:

+- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, true] as bigint)),false), [id=#2183]
  +- BroadcastQueryStage 2
    +- ReusedExchange [d_date_sk#1086], BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, true] as bigint)),false), [id=#1963]

A root cause is that a Cast class in a required child's distribution does not have a timeZoneId field (timeZoneId=None), and a Cast class in child.outputPartitioning has it. So, this difference can make the distribution requirement check fail in EnsureRequirements:

spark/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala

Lines 47 to 50 in 1e85707

    
           case (child, distribution) if child.outputPartitioning.satisfies(distribution) => 
        
             child 
        
           case (child, BroadcastDistribution(mode)) => 
        
             BroadcastExchangeExec(mode, child)

The Cast class that does not have a timeZoneId field is generated in the HashJoin object. To fix this issue, this PR proposes to use the CastSupport.cast method there.

Why are the changes needed?

Bugfix.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Manually checked that q5 passed.

maropu · 2020-12-17T07:02:07Z

To add unit tests, I'm looking for a simpler query to reproduce this. But, I've not found it yet.

maropu · 2020-12-17T07:07:52Z

cc: @dongjoon-hyun @cloud-fan @HyukjinKwon

dongjoon-hyun

Wow, super quick! @maropu :)

dongjoon-hyun

+1, LGTM. Thank you, @maropu . I manually verified that this PR fixes TPCDS Q5 error.

dongjoon-hyun · 2020-12-17T07:34:50Z

BTW, @maropu . Why do you make another JIRA for this? Also, the new JIRA seems to be an Improvement? Do we need more patch to handle Q5?

If you don't mind, shall we use the original issue, SPARK-33822 (Bug) instead of SPARK-33823 (Improvement)?

HyukjinKwon · 2020-12-17T07:53:54Z

Yeah let's port it back.

SparkQA · 2020-12-17T07:58:21Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37540/

dongjoon-hyun · 2020-12-17T07:58:30Z

Thanks, I merged two issues by copying all description into one JIRA (SPARK-33822 Bug).

maropu · 2020-12-17T08:02:23Z

BTW, @maropu . Why do you make another JIRA for this? Also, the new JIRA seems to be an Improvement? Do we need more patch to handle Q5?

Yea, it's just my mistake... (I didn't notice that you filed the ticket for this).

SparkQA · 2020-12-17T08:27:29Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37540/

cloud-fan

good catch!

SparkQA · 2020-12-17T08:56:21Z

Test build #132937 has finished for PR 30818 at commit 7e33df3.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

HyukjinKwon · 2020-12-17T09:03:57Z

retest this please

gengliangwang

Thanks for the quick fix!

SparkQA · 2020-12-17T11:51:37Z

Test build #132943 has finished for PR 30818 at commit 7e33df3.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

dongjoon-hyun · 2020-12-17T17:20:17Z

Could you fix the test suite, @maropu ?

The following is failing due to the session timezone.

    assert(HashJoin.rewriteKeyExpr(i :: Nil) === Cast(i, LongType) :: Nil)

viirya

nice catch!

maropu · 2020-12-17T22:15:52Z

Sure, @dongjoon-hyun

dongjoon-hyun · 2020-12-18T00:13:48Z

Thank you, @maropu and all. The last commit is only about the two test suites and I verified them locally.
Merged to master/3.1/3.0.

maropu · 2020-12-18T00:15:16Z

Thanks for the quick response, @dongjoon-hyun !

### What changes were proposed in this pull request? This PR intends to fix the bug that throws a unsupported exception when running [the TPCDS q5](https://github.com/apache/spark/blob/master/sql/core/src/test/resources/tpcds/q5.sql) with AQE enabled ([this option is enabled by default now via SPARK-33679](031c5ef)): ``` java.lang.UnsupportedOperationException: BroadcastExchange does not support the execute() code path. at org.apache.spark.sql.execution.exchange.BroadcastExchangeExec.doExecute(BroadcastExchangeExec.scala:189) at org.apache.spark.sql.execution.SparkPlan.$anonfun$execute$1(SparkPlan.scala:180) at org.apache.spark.sql.execution.SparkPlan.$anonfun$executeQuery$1(SparkPlan.scala:218) at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151) at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:215) at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:176) at org.apache.spark.sql.execution.exchange.ReusedExchangeExec.doExecute(Exchange.scala:60) at org.apache.spark.sql.execution.SparkPlan.$anonfun$execute$1(SparkPlan.scala:180) at org.apache.spark.sql.execution.SparkPlan.$anonfun$executeQuery$1(SparkPlan.scala:218) at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151) at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:215) at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:176) at org.apache.spark.sql.execution.adaptive.QueryStageExec.doExecute(QueryStageExec.scala:115) at org.apache.spark.sql.execution.SparkPlan.$anonfun$execute$1(SparkPlan.scala:180) at org.apache.spark.sql.execution.SparkPlan.$anonfun$executeQuery$1(SparkPlan.scala:218) at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151) at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:215) at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:176) at org.apache.spark.sql.execution.SparkPlan.getByteArrayRdd(SparkPlan.scala:321) at org.apache.spark.sql.execution.SparkPlan.executeCollectIterator(SparkPlan.scala:397) at org.apache.spark.sql.execution.exchange.BroadcastExchangeExec.$anonfun$relationFuture$1(BroadcastExchangeExec.scala:118) at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withThreadLocalCaptured$1(SQLExecution.scala:185) at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) ... ``` I've checked the AQE code and I found `EnsureRequirements` wrongly puts `BroadcastExchange` on a top of `BroadcastQueryStage` in the `reOptimize` phase as follows: ``` +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, true] as bigint)),false), [id=#2183] +- BroadcastQueryStage 2 +- ReusedExchange [d_date_sk#1086], BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, true] as bigint)),false), [id=#1963] ``` A root cause is that a `Cast` class in a required child's distribution does not have a `timeZoneId` field (`timeZoneId=None`), and a `Cast` class in `child.outputPartitioning` has it. So, this difference can make the distribution requirement check fail in `EnsureRequirements`: https://github.com/apache/spark/blob/1e85707738a830d33598ca267a6740b3f06b1861/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala#L47-L50 The `Cast` class that does not have a `timeZoneId` field is generated in the `HashJoin` object. To fix this issue, this PR proposes to use the `CastSupport.cast` method there. ### Why are the changes needed? Bugfix. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Manually checked that q5 passed. Closes #30818 from maropu/BugfixInAQE. Authored-by: Takeshi Yamamuro <yamamuro@apache.org> Signed-off-by: Dongjoon Hyun <dongjoon@apache.org> (cherry picked from commit 51ef443) Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

dongjoon-hyun · 2020-12-18T00:39:55Z

Ur, I forgot that SQLConfHelper doesn't exist in branch-3.0.
Could you make a backport PR to branch-3.0, please? @maropu ?

maropu · 2020-12-18T00:59:06Z

okay, I will do it now.

gatorsmile · 2020-12-18T02:16:37Z

Thank you! @maropu

### What changes were proposed in this pull request? This PR intends to fix the bug that throws a unsupported exception when running [the TPCDS q5](https://github.com/apache/spark/blob/master/sql/core/src/test/resources/tpcds/q5.sql) with AQE enabled ([this option is enabled by default now via SPARK-33679](031c5ef)): ``` java.lang.UnsupportedOperationException: BroadcastExchange does not support the execute() code path. at org.apache.spark.sql.execution.exchange.BroadcastExchangeExec.doExecute(BroadcastExchangeExec.scala:189) at org.apache.spark.sql.execution.SparkPlan.$anonfun$execute$1(SparkPlan.scala:180) at org.apache.spark.sql.execution.SparkPlan.$anonfun$executeQuery$1(SparkPlan.scala:218) at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151) at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:215) at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:176) at org.apache.spark.sql.execution.exchange.ReusedExchangeExec.doExecute(Exchange.scala:60) at org.apache.spark.sql.execution.SparkPlan.$anonfun$execute$1(SparkPlan.scala:180) at org.apache.spark.sql.execution.SparkPlan.$anonfun$executeQuery$1(SparkPlan.scala:218) at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151) at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:215) at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:176) at org.apache.spark.sql.execution.adaptive.QueryStageExec.doExecute(QueryStageExec.scala:115) at org.apache.spark.sql.execution.SparkPlan.$anonfun$execute$1(SparkPlan.scala:180) at org.apache.spark.sql.execution.SparkPlan.$anonfun$executeQuery$1(SparkPlan.scala:218) at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151) at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:215) at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:176) at org.apache.spark.sql.execution.SparkPlan.getByteArrayRdd(SparkPlan.scala:321) at org.apache.spark.sql.execution.SparkPlan.executeCollectIterator(SparkPlan.scala:397) at org.apache.spark.sql.execution.exchange.BroadcastExchangeExec.$anonfun$relationFuture$1(BroadcastExchangeExec.scala:118) at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withThreadLocalCaptured$1(SQLExecution.scala:185) at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) ... ``` I've checked the AQE code and I found `EnsureRequirements` wrongly puts `BroadcastExchange` on a top of `BroadcastQueryStage` in the `reOptimize` phase as follows: ``` +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, true] as bigint)),false), [id=#2183] +- BroadcastQueryStage 2 +- ReusedExchange [d_date_sk#1086], BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, true] as bigint)),false), [id=#1963] ``` A root cause is that a `Cast` class in a required child's distribution does not have a `timeZoneId` field (`timeZoneId=None`), and a `Cast` class in `child.outputPartitioning` has it. So, this difference can make the distribution requirement check fail in `EnsureRequirements`: https://github.com/apache/spark/blob/1e85707738a830d33598ca267a6740b3f06b1861/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala#L47-L50 The `Cast` class that does not have a `timeZoneId` field is generated in the `HashJoin` object. To fix this issue, this PR proposes to use the `CastSupport.cast` method there. This is a backport PR for #30818. ### Why are the changes needed? Bugfix. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Manually checked that q5 passed with AQE enabled. Closes #30830 from maropu/SPARK-33822-BRANCH3.0. Authored-by: Takeshi Yamamuro <yamamuro@apache.org> Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

LeonidVasilyev · 2024-04-16T13:23:06Z

Could you please tell if this has been fixed in 3.2 and the newer versions as well? The ticket lists 3.0.0, 3.0.1, 3.1.0, 3.2.0 as the affected versions and 3.0.2, 3.1.0 as fix versions. I believe we're having this issue in 3.2.1.

cloud-fan · 2024-04-16T13:49:32Z

It's fixed in 3.1.0 and forward, including 3.2.0. You may hit a different bug, please open a new JIRA ticket, thanks for reporting!

Fix

7e33df3

dongjoon-hyun reviewed Dec 17, 2020

View reviewed changes

dongjoon-hyun approved these changes Dec 17, 2020

View reviewed changes

github-actions bot added the SQL label Dec 17, 2020

dongjoon-hyun changed the title ~~[SPARK-33823][SQL] Use the CastSupport.cast method in HashJoin~~ [SPARK-33822][SQL] Use the CastSupport.cast method in HashJoin Dec 17, 2020

cloud-fan approved these changes Dec 17, 2020

View reviewed changes

gengliangwang approved these changes Dec 17, 2020

View reviewed changes

viirya approved these changes Dec 17, 2020

View reviewed changes

viirya reviewed Dec 17, 2020

View reviewed changes

test fixes

0739fda

dongjoon-hyun closed this in 51ef443 Dec 18, 2020

maropu mentioned this pull request Dec 18, 2020

[SPARK-33822][SQL][3.0] Use the CastSupport.cast method in HashJoin #30830

Closed

	case (child, distribution) if child.outputPartitioning.satisfies(distribution) =>
	child
	case (child, BroadcastDistribution(mode)) =>
	BroadcastExchangeExec(mode, child)

[SPARK-33822][SQL] Use the CastSupport.cast method in HashJoin #30818

[SPARK-33822][SQL] Use the CastSupport.cast method in HashJoin #30818

Uh oh!

Conversation

maropu commented Dec 17, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

maropu commented Dec 17, 2020

Uh oh!

maropu commented Dec 17, 2020

Uh oh!

dongjoon-hyun left a comment

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun left a comment

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun commented Dec 17, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

HyukjinKwon commented Dec 17, 2020

Uh oh!

SparkQA commented Dec 17, 2020

Uh oh!

dongjoon-hyun commented Dec 17, 2020

Uh oh!

maropu commented Dec 17, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

SparkQA commented Dec 17, 2020

Uh oh!

cloud-fan left a comment

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Dec 17, 2020

Uh oh!

HyukjinKwon commented Dec 17, 2020

Uh oh!

gengliangwang left a comment

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Dec 17, 2020

Uh oh!

dongjoon-hyun commented Dec 17, 2020

Uh oh!

viirya left a comment

Choose a reason for hiding this comment

Uh oh!

maropu commented Dec 17, 2020

Uh oh!

dongjoon-hyun commented Dec 18, 2020

Uh oh!

maropu commented Dec 18, 2020

Uh oh!

dongjoon-hyun commented Dec 18, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

maropu commented Dec 18, 2020

Uh oh!

gatorsmile commented Dec 18, 2020

Uh oh!

LeonidVasilyev commented Apr 16, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

cloud-fan commented Apr 16, 2024

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

9 participants

[SPARK-33822][SQL] Use the `CastSupport.cast` method in HashJoin #30818

[SPARK-33822][SQL] Use the `CastSupport.cast` method in HashJoin #30818

maropu commented Dec 17, 2020 •

edited

Loading

dongjoon-hyun commented Dec 17, 2020 •

edited

Loading

maropu commented Dec 17, 2020 •

edited

Loading

dongjoon-hyun commented Dec 18, 2020 •

edited

Loading

LeonidVasilyev commented Apr 16, 2024 •

edited

Loading