#28 ignore some ut, consider replacing the spark version to solve #34
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
7mming7
pushed a commit
that referenced
this pull request
Nov 4, 2020
### What changes were proposed in this pull request?
To support formatted explain for AQE.
### Why are the changes needed?
AQE does not support formatted explain yet. It's good to support it for better user experience, debugging, etc.
Before:
```
== Physical Plan ==
AdaptiveSparkPlan (1)
+- * HashAggregate (unknown)
+- CustomShuffleReader (unknown)
+- ShuffleQueryStage (unknown)
+- Exchange (unknown)
+- * HashAggregate (unknown)
+- * Project (unknown)
+- * BroadcastHashJoin Inner BuildRight (unknown)
:- * LocalTableScan (unknown)
+- BroadcastQueryStage (unknown)
+- BroadcastExchange (unknown)
+- LocalTableScan (unknown)
(1) AdaptiveSparkPlan
Output [4]: [k#7, count(v1)#32L, sum(v1)#33L, avg(v2)#34]
Arguments: HashAggregate(keys=[k#7], functions=[count(1), sum(cast(v1#8 as bigint)), avg(cast(v2#19 as bigint))]), AdaptiveExecutionContext(org.apache.spark.sql.SparkSession104ab57b), [PlanAdaptiveSubqueries(Map())], false
```
After:
```
== Physical Plan ==
AdaptiveSparkPlan (14)
+- * HashAggregate (13)
+- CustomShuffleReader (12)
+- ShuffleQueryStage (11)
+- Exchange (10)
+- * HashAggregate (9)
+- * Project (8)
+- * BroadcastHashJoin Inner BuildRight (7)
:- * Project (2)
: +- * LocalTableScan (1)
+- BroadcastQueryStage (6)
+- BroadcastExchange (5)
+- * Project (4)
+- * LocalTableScan (3)
(1) LocalTableScan [codegen id : 2]
Output [2]: [_1#x, _2#x]
Arguments: [_1#x, _2#x]
(2) Project [codegen id : 2]
Output [2]: [_1#x AS k#x, _2#x AS v1#x]
Input [2]: [_1#x, _2#x]
(3) LocalTableScan [codegen id : 1]
Output [2]: [_1#x, _2#x]
Arguments: [_1#x, _2#x]
(4) Project [codegen id : 1]
Output [2]: [_1#x AS k#x, _2#x AS v2#x]
Input [2]: [_1#x, _2#x]
(5) BroadcastExchange
Input [2]: [k#x, v2#x]
Arguments: HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint))), [id=#x]
(6) BroadcastQueryStage
Output [2]: [k#x, v2#x]
Arguments: 0
(7) BroadcastHashJoin [codegen id : 2]
Left keys [1]: [k#x]
Right keys [1]: [k#x]
Join condition: None
(8) Project [codegen id : 2]
Output [3]: [k#x, v1#x, v2#x]
Input [4]: [k#x, v1#x, k#x, v2#x]
(9) HashAggregate [codegen id : 2]
Input [3]: [k#x, v1#x, v2#x]
Keys [1]: [k#x]
Functions [3]: [partial_count(1), partial_sum(cast(v1#x as bigint)), partial_avg(cast(v2#x as bigint))]
Aggregate Attributes [4]: [count#xL, sum#xL, sum#x, count#xL]
Results [5]: [k#x, count#xL, sum#xL, sum#x, count#xL]
(10) Exchange
Input [5]: [k#x, count#xL, sum#xL, sum#x, count#xL]
Arguments: hashpartitioning(k#x, 5), true, [id=#x]
(11) ShuffleQueryStage
Output [5]: [sum#xL, k#x, sum#x, count#xL, count#xL]
Arguments: 1
(12) CustomShuffleReader
Input [5]: [k#x, count#xL, sum#xL, sum#x, count#xL]
Arguments: coalesced
(13) HashAggregate [codegen id : 3]
Input [5]: [k#x, count#xL, sum#xL, sum#x, count#xL]
Keys [1]: [k#x]
Functions [3]: [count(1), sum(cast(v1#x as bigint)), avg(cast(v2#x as bigint))]
Aggregate Attributes [3]: [count(1)#xL, sum(cast(v1#x as bigint))#xL, avg(cast(v2#x as bigint))#x]
Results [4]: [k#x, count(1)#xL AS count(v1)#xL, sum(cast(v1#x as bigint))#xL AS sum(v1)#xL, avg(cast(v2#x as bigint))#x AS avg(v2)#x]
(14) AdaptiveSparkPlan
Output [4]: [k#x, count(v1)#xL, sum(v1)#xL, avg(v2)#x]
Arguments: isFinalPlan=true
```
### Does this PR introduce any user-facing change?
No, this should be new feature along with AQE in Spark 3.0.
### How was this patch tested?
Added a query file: `explain-aqe.sql` and a unit test.
Closes apache#28271 from Ngone51/support_formatted_explain_for_aqe.
Authored-by: yi.wu <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
7mming7
pushed a commit
that referenced
this pull request
Nov 4, 2020
…or its output partitioning
### What changes were proposed in this pull request?
Currently, the `BroadcastHashJoinExec`'s `outputPartitioning` only uses the streamed side's `outputPartitioning`. However, if the join type of `BroadcastHashJoinExec` is an inner-like join, the build side's info (the join keys) can be added to `BroadcastHashJoinExec`'s `outputPartitioning`.
For example,
```Scala
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "500")
val t1 = (0 until 100).map(i => (i % 5, i % 13)).toDF("i1", "j1")
val t2 = (0 until 100).map(i => (i % 5, i % 13)).toDF("i2", "j2")
val t3 = (0 until 20).map(i => (i % 7, i % 11)).toDF("i3", "j3")
val t4 = (0 until 100).map(i => (i % 5, i % 13)).toDF("i4", "j4")
// join1 is a sort merge join.
val join1 = t1.join(t2, t1("i1") === t2("i2"))
// join2 is a broadcast join where t3 is broadcasted.
val join2 = join1.join(t3, join1("i1") === t3("i3"))
// Join on the column from the broadcasted side (i3).
val join3 = join2.join(t4, join2("i3") === t4("i4"))
join3.explain
```
You see that `Exchange hashpartitioning(i2#103, 200)` is introduced because there is no output partitioning info from the build side.
```
== Physical Plan ==
*(6) SortMergeJoin [i3#29], [i4#40], Inner
:- *(4) Sort [i3#29 ASC NULLS FIRST], false, 0
: +- Exchange hashpartitioning(i3#29, 200), true, [id=#55]
: +- *(3) BroadcastHashJoin [i1#7], [i3#29], Inner, BuildRight
: :- *(3) SortMergeJoin [i1#7], [i2#18], Inner
: : :- *(1) Sort [i1#7 ASC NULLS FIRST], false, 0
: : : +- Exchange hashpartitioning(i1#7, 200), true, [id=#28]
: : : +- LocalTableScan [i1#7, j1#8]
: : +- *(2) Sort [i2#18 ASC NULLS FIRST], false, 0
: : +- Exchange hashpartitioning(i2#18, 200), true, [id=#29]
: : +- LocalTableScan [i2#18, j2#19]
: +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint))), [id=#34]
: +- LocalTableScan [i3#29, j3#30]
+- *(5) Sort [i4#40 ASC NULLS FIRST], false, 0
+- Exchange hashpartitioning(i4#40, 200), true, [id=#39]
+- LocalTableScan [i4#40, j4#41]
```
This PR proposes to introduce output partitioning for the build side for `BroadcastHashJoinExec` if the streamed side has a `HashPartitioning` or a collection of `HashPartitioning`s.
There is a new internal config `spark.sql.execution.broadcastHashJoin.outputPartitioningExpandLimit`, which can limit the number of partitioning a `HashPartitioning` can expand to. It can be set to "0" to disable this feature.
### Why are the changes needed?
To remove unnecessary shuffle.
### Does this PR introduce _any_ user-facing change?
Yes, now the shuffle in the above example can be eliminated:
```
== Physical Plan ==
*(5) SortMergeJoin [i3#108], [i4#119], Inner
:- *(3) Sort [i3#108 ASC NULLS FIRST], false, 0
: +- *(3) BroadcastHashJoin [i1#86], [i3#108], Inner, BuildRight
: :- *(3) SortMergeJoin [i1#86], [i2#97], Inner
: : :- *(1) Sort [i1#86 ASC NULLS FIRST], false, 0
: : : +- Exchange hashpartitioning(i1#86, 200), true, [id=#120]
: : : +- LocalTableScan [i1#86, j1#87]
: : +- *(2) Sort [i2#97 ASC NULLS FIRST], false, 0
: : +- Exchange hashpartitioning(i2#97, 200), true, [id=#121]
: : +- LocalTableScan [i2#97, j2#98]
: +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint))), [id=#126]
: +- LocalTableScan [i3#108, j3#109]
+- *(4) Sort [i4#119 ASC NULLS FIRST], false, 0
+- Exchange hashpartitioning(i4#119, 200), true, [id=#130]
+- LocalTableScan [i4#119, j4#120]
```
### How was this patch tested?
Added new tests.
Closes apache#28676 from imback82/broadcast_join_output.
Authored-by: Terry Kim <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
chenzhx
pushed a commit
to chenzhx/spark
that referenced
this pull request
Jan 26, 2022
…on to make Java 17 compatible with Java 8
### What changes were proposed in this pull request?
The `date_format` function with `B` format has different behavior when use Java 8 and Java 17, `select date_format('2018-11-17 13:33:33.333', 'B')` in `datetime-formatting-invalid.sql` can prove this.
The case result with Java 8 is
```
-- !query
select date_format('2018-11-17 13:33:33.333', 'B')
-- !query schema
struct<>
-- !query output
java.lang.IllegalArgumentException
Unknown pattern letter: B
```
and the case result with Java 17 is
```
- datetime-formatting-invalid.sql *** FAILED ***
datetime-formatting-invalid.sql
Expected "struct<[]>", but got "struct<[date_format(2018-11-17 13:33:33.333, B):string]>" Schema did not match for query Kyligence#34
select date_format('2018-11-17 13:33:33.333', 'B'): -- !query
select date_format('2018-11-17 13:33:33.333', 'B')
-- !query schema
struct<date_format(2018-11-17 13:33:33.333, B):string>
-- !query output
in the afternoon (SQLQueryTestSuite.scala:469)
```
We found that this is due to the new support of format `B` in Java 17
```
'B' is used to represent Pattern letters to output a day period in Java 17
* Pattern Count Equivalent builder methods
* ------- ----- --------------------------
* B 1 appendDayPeriodText(TextStyle.SHORT)
* BBBB 4 appendDayPeriodText(TextStyle.FULL)
* BBBBB 5 appendDayPeriodText(TextStyle.NARROW)
```
And through [ http://spark.apache.org/docs/latest/sql-ref-datetime-pattern.html]( http://spark.apache.org/docs/latest/sql-ref-datetime-pattern.html) , we can confirm that format `B` is not documented/supported for `date_format` function currently.
So the main change of this pr is manual disabled format `B` of `date_format` function in `DateTimeFormatterHelper` to make Java 17 compatible with Java 8.
### Why are the changes needed?
Ensure that Java 17 and Java 8 have the same behavior.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
- Pass the Jenkins or GitHub Action
- Manual test `SQLQueryTestSuite` with JDK 17
**Before**
```
- datetime-formatting-invalid.sql *** FAILED ***
datetime-formatting-invalid.sql
Expected "struct<[]>", but got "struct<[date_format(2018-11-17 13:33:33.333, B):string]>" Schema did not match for query Kyligence#34
select date_format('2018-11-17 13:33:33.333', 'B'): -- !query
select date_format('2018-11-17 13:33:33.333', 'B')
-- !query schema
struct<date_format(2018-11-17 13:33:33.333, B):string>
-- !query output
in the afternoon (SQLQueryTestSuite.scala:469)
```
**After**
The test `select date_format('2018-11-17 13:33:33.333', 'B')` in `datetime-formatting-invalid.sql` passed
Closes apache#34237 from LuciferYang/SPARK-36970.
Authored-by: yangjie01 <[email protected]>
Signed-off-by: Max Gekk <[email protected]>
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
1 ignore some ut, consider replacing the spark version to solve
2 change spark download url to speed up