You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2020/12/18 02:47:15 UTC
[spark] branch branch-3.0 updated: [SPARK-33822][SQL][3.0] Use the
`CastSupport.cast` method in HashJoin
This is an automated email from the ASF dual-hosted git repository.
dongjoon pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.0 by this push:
new 1615b0e [SPARK-33822][SQL][3.0] Use the `CastSupport.cast` method in HashJoin
1615b0e is described below
commit 1615b0ec28a316b04c31e2ae4e54495c105f7cf2
Author: Takeshi Yamamuro <ya...@apache.org>
AuthorDate: Thu Dec 17 18:43:21 2020 -0800
[SPARK-33822][SQL][3.0] Use the `CastSupport.cast` method in HashJoin
### 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](https://github.com/apache/spark/commit/031c5ef280e0cba8c4718a6457a44b6cccb17f46)):
```
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 <ya...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../spark/sql/execution/joins/HashJoin.scala | 15 ++++++----
.../sql/execution/joins/BroadcastJoinSuite.scala | 33 +++++++++++++---------
2 files changed, 30 insertions(+), 18 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashJoin.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashJoin.scala
index 931a75a..0275741 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashJoin.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashJoin.scala
@@ -18,12 +18,14 @@
package org.apache.spark.sql.execution.joins
import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.CastSupport
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.BindReferences.bindReferences
import org.apache.spark.sql.catalyst.plans._
import org.apache.spark.sql.catalyst.plans.physical.Partitioning
import org.apache.spark.sql.execution.{ExplainUtils, RowIterator, SparkPlan}
import org.apache.spark.sql.execution.metric.SQLMetric
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{IntegralType, LongType}
trait HashJoin {
@@ -240,7 +242,10 @@ trait HashJoin {
}
}
-object HashJoin {
+object HashJoin extends CastSupport {
+
+ override def conf: SQLConf = SQLConf.get
+
/**
* Try to rewrite the key as LongType so we can use getLong(), if they key can fit with a long.
*
@@ -255,14 +260,14 @@ object HashJoin {
}
var keyExpr: Expression = if (keys.head.dataType != LongType) {
- Cast(keys.head, LongType)
+ cast(keys.head, LongType)
} else {
keys.head
}
keys.tail.foreach { e =>
val bits = e.dataType.defaultSize * 8
keyExpr = BitwiseOr(ShiftLeft(keyExpr, Literal(bits)),
- BitwiseAnd(Cast(e, LongType), Literal((1L << bits) - 1)))
+ BitwiseAnd(cast(e, LongType), Literal((1L << bits) - 1)))
}
keyExpr :: Nil
}
@@ -275,13 +280,13 @@ object HashJoin {
// jump over keys that have a higher index value than the required key
if (keys.size == 1) {
assert(index == 0)
- Cast(BoundReference(0, LongType, nullable = false), keys(index).dataType)
+ cast(BoundReference(0, LongType, nullable = false), keys(index).dataType)
} else {
val shiftedBits =
keys.slice(index + 1, keys.size).map(_.dataType.defaultSize * 8).sum
val mask = (1L << (keys(index).dataType.defaultSize * 8)) - 1
// build the schema for unpacking the required key
- Cast(BitwiseAnd(
+ cast(BitwiseAnd(
ShiftRightUnsigned(BoundReference(0, LongType, nullable = false), Literal(shiftedBits)),
Literal(mask)), keys(index).dataType)
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/joins/BroadcastJoinSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/joins/BroadcastJoinSuite.scala
index 335ef25..ef0a596 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/joins/BroadcastJoinSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/joins/BroadcastJoinSuite.scala
@@ -239,33 +239,40 @@ abstract class BroadcastJoinSuiteBase extends QueryTest with SQLTestUtils
assert(HashJoin.rewriteKeyExpr(l :: l :: Nil) === l :: l :: Nil)
assert(HashJoin.rewriteKeyExpr(l :: i :: Nil) === l :: i :: Nil)
- assert(HashJoin.rewriteKeyExpr(i :: Nil) === Cast(i, LongType) :: Nil)
+ assert(HashJoin.rewriteKeyExpr(i :: Nil) ===
+ Cast(i, LongType, Some(conf.sessionLocalTimeZone)) :: Nil)
assert(HashJoin.rewriteKeyExpr(i :: l :: Nil) === i :: l :: Nil)
assert(HashJoin.rewriteKeyExpr(i :: i :: Nil) ===
- BitwiseOr(ShiftLeft(Cast(i, LongType), Literal(32)),
- BitwiseAnd(Cast(i, LongType), Literal((1L << 32) - 1))) :: Nil)
+ BitwiseOr(ShiftLeft(Cast(i, LongType, Some(conf.sessionLocalTimeZone)), Literal(32)),
+ BitwiseAnd(Cast(i, LongType, Some(conf.sessionLocalTimeZone)), Literal((1L << 32) - 1))) ::
+ Nil)
assert(HashJoin.rewriteKeyExpr(i :: i :: i :: Nil) === i :: i :: i :: Nil)
- assert(HashJoin.rewriteKeyExpr(s :: Nil) === Cast(s, LongType) :: Nil)
+ assert(HashJoin.rewriteKeyExpr(s :: Nil) ===
+ Cast(s, LongType, Some(conf.sessionLocalTimeZone)) :: Nil)
assert(HashJoin.rewriteKeyExpr(s :: l :: Nil) === s :: l :: Nil)
assert(HashJoin.rewriteKeyExpr(s :: s :: Nil) ===
- BitwiseOr(ShiftLeft(Cast(s, LongType), Literal(16)),
- BitwiseAnd(Cast(s, LongType), Literal((1L << 16) - 1))) :: Nil)
+ BitwiseOr(ShiftLeft(Cast(s, LongType, Some(conf.sessionLocalTimeZone)), Literal(16)),
+ BitwiseAnd(Cast(s, LongType, Some(conf.sessionLocalTimeZone)), Literal((1L << 16) - 1))) ::
+ Nil)
assert(HashJoin.rewriteKeyExpr(s :: s :: s :: Nil) ===
BitwiseOr(ShiftLeft(
- BitwiseOr(ShiftLeft(Cast(s, LongType), Literal(16)),
- BitwiseAnd(Cast(s, LongType), Literal((1L << 16) - 1))),
+ BitwiseOr(ShiftLeft(Cast(s, LongType, Some(conf.sessionLocalTimeZone)), Literal(16)),
+ BitwiseAnd(Cast(s, LongType, Some(conf.sessionLocalTimeZone)), Literal((1L << 16) - 1))),
Literal(16)),
- BitwiseAnd(Cast(s, LongType), Literal((1L << 16) - 1))) :: Nil)
+ BitwiseAnd(Cast(s, LongType, Some(conf.sessionLocalTimeZone)), Literal((1L << 16) - 1))) ::
+ Nil)
assert(HashJoin.rewriteKeyExpr(s :: s :: s :: s :: Nil) ===
BitwiseOr(ShiftLeft(
BitwiseOr(ShiftLeft(
- BitwiseOr(ShiftLeft(Cast(s, LongType), Literal(16)),
- BitwiseAnd(Cast(s, LongType), Literal((1L << 16) - 1))),
+ BitwiseOr(ShiftLeft(Cast(s, LongType, Some(conf.sessionLocalTimeZone)), Literal(16)),
+ BitwiseAnd(Cast(s, LongType, Some(conf.sessionLocalTimeZone)),
+ Literal((1L << 16) - 1))),
Literal(16)),
- BitwiseAnd(Cast(s, LongType), Literal((1L << 16) - 1))),
+ BitwiseAnd(Cast(s, LongType, Some(conf.sessionLocalTimeZone)), Literal((1L << 16) - 1))),
Literal(16)),
- BitwiseAnd(Cast(s, LongType), Literal((1L << 16) - 1))) :: Nil)
+ BitwiseAnd(Cast(s, LongType, Some(conf.sessionLocalTimeZone)), Literal((1L << 16) - 1))) ::
+ Nil)
assert(HashJoin.rewriteKeyExpr(s :: s :: s :: s :: s :: Nil) ===
s :: s :: s :: s :: s :: Nil)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org