[SPARK-37193][SQL] DynamicJoinSelection.shouldDemoteBroadcastHashJoin should not apply to outer joins #34464

ekoifman · 2021-11-02T03:40:42Z

What changes were proposed in this pull request?

DynamicJoinSelection.shouldDemoteBroadcastHashJoin will prevent AQE from converting Sort merge join into a broadcast join because SMJ is faster when the side that would be broadcast has a lot of empty partitions.
This makes sense for inner joins which can short circuit if one side is empty.
For (left,right) outer join, the streaming side still has to be processed so demoting broadcast join doesn't have the same advantage.

Does this PR introduce any user-facing change?

Yes, it may cause AQE to choose BHJ more often than before with better performance

How was this patch tested?

Spark UTs
Also empirical evidence

AmplabJenkins · 2021-11-02T03:41:19Z

Can one of the admins verify this patch?

sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala

singhpk234 · 2021-11-02T08:50:40Z

sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/DynamicJoinSelection.scala

+      val demoteBroadcastHash = joinType match {
+        // doesn't make sense for outer joins since one side is preserved and join is not
+        // short circuited if the other side is empty
+        case Inner | LeftSemi | LeftAnti => shouldDemoteBroadcastHashJoin (stage.mapStats.get)


Can we also include Existence Joins as well

LeftExistence pattern to match all LeftSemi / LeftAnti / Existence in one shot

changed it a bit - seems more natural

sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/DynamicJoinSelection.scala

singhpk234 · 2021-11-02T09:01:47Z

sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala

+      withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
+        val (plan, adaptivePlan) = runAdaptiveAndVerifyResult(
+          "SELECT * FROM (select * from testData where value = '1') td" +
+            " right outer join testData2 ON key = a")


[optional] how about adding all outerJoinTypes here and making it more rigid
can be done with minimal changes

Seq("right outer", "left outer", "full outer").foreach { joinType => val (plan, adaptivePlan) = runAdaptiveAndVerifyResult( "SELECT * FROM (select * from testData where value = '1') td" + s" $joinType join testData2 ON key = a") val smj = findTopLevelSortMergeJoin(plan) assert(smj.size == 1) val bhj = findTopLevelBroadcastHashJoin(adaptivePlan) assert(bhj.size == 1) }

Thanks for the feedback but changing right outer here to a left outer won't have the same semantics w/o changing the sql. You can only broadcast the inner side.
I don't believe full outer supports BHJ at all

sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/DynamicJoinSelection.scala

ekoifman · 2021-11-10T14:59:50Z

@HyukjinKwon do you have any other suggestions?

ekoifman · 2021-11-15T19:05:25Z

Hi
@cloud-fan @ulysses-you @maryannxue could one of you take a look please

ulysses-you · 2021-11-16T09:55:22Z

sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/DynamicJoinSelection.scala

+      val demoteBroadcastHash = joinType match {
+        // doesn't make sense for outer joins since one side is preserved and join is not
+        // short circuited if the other side is empty
+        case LeftOuter | RightOuter | FullOuter => false


does BHJ support FullOuter ? what about LeftSemi and LeftAnti ?

BHJ can't support FullOuter - I put it in the list since it seems odd (though not wrong) to add no_broadcast_hash to a join that doesn't support BHJ anyway. Makes logs harder to read.
LeftSemi & LeftAnti could in principle support it - I don't know if they do today.

ekoifman · 2021-12-21T18:43:47Z

ping - is there anything else I can do to move this along?

cloud-fan · 2021-12-22T10:25:07Z

The last tests pass is long time ago, @ekoifman can you rebase this PR and run the test again? Sorry for the late review.

… should not apply to outer joins

ekoifman · 2021-12-22T19:37:25Z

@cloud-fan done

cloud-fan · 2021-12-23T15:06:29Z

thanks, merging to master!

cloud-fan · 2021-12-23T15:11:20Z

sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/DynamicJoinSelection.scala

      var newHint = hint
      if (!hint.leftHint.exists(_.strategy.isDefined)) {
-        selectJoinStrategy(left).foreach { strategy =>
+        selectJoinStrategy(left, joinType).foreach { strategy =>


For left outer join, and the left side has many empty partitions, shall we still demote?

For LOJ with many empty partitions on the left, the local join can short-circuit whether you broadcast or shuffle. I'm not sure how to determine which strategy will send less data around. Is there another heuristic that can be used?

the local join can short-circuit whether you broadcast or shuffle.

I think there is a misunderstanding here. We see many empty partitions on the shuffled left side, it doesn't mean the original left side before shuffle also has many empty partitions. I think we need to demote.

@ekoifman can you open a followup PR to fix this issue?

You're right. filed SPARK-37753

cloud-fan · 2022-01-07T14:15:37Z

I'm reverting this to avoid mistakenly releasing a performance regression in Spark 3.3. Please resubmit this PR with SPARK-37753 resolved.

…ynamicJoinSelection ### What changes were proposed in this pull request? ### Why are the changes needed? In the current implementation of DynamicJoinSelection the logic checks if one side of the join has high ratio of empty partitions and adds a NO_BROADCAST hint on that side since a shuffle join can short-circuit the local joins where one side is empty. This logic is doesn't make sense for all join type. For example, a Left Outer Join cannot short circuit if RHS is empty so we should not inhibit BHJ. On the other hand a LOJ executed as a shuffle join where the LHS has many empty can short circuit the local join so we should inhibit the BHJ because BHJ will use OptimizeShuffleWithLocalRead which will re-assemble LHS partitions as the were before the shuffle and thus may not have many empty ones any more. This supersedes [SPARK-37193](https://issues.apache.org/jira/browse/SPARK-37193) Also see previous discussion in #34464 (comment) ### Does this PR introduce _any_ user-facing change? It may change which joins run as BHJ vs shuffle joins ### How was this patch tested? Unit Tests Closes #35715 from ekoifman/SPARK-37753-enhance-DynamiJoinSelection. Authored-by: Eugene Koifman <eugene.koifman@workday.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>

thomasg19930417 · 2022-09-15T10:01:07Z

I'm reverting this to avoid mistakenly releasing a performance regression in Spark 3.3. Please resubmit this PR with SPARK-37753 resolved.

can you tell me ,What does short circuit local join mean,thanks

ekoifman · 2022-09-15T15:25:37Z

@thomasg19930417
Imagine you have an inner join where the one side has 0 rows. As soon as you know that one side is empty, you don't have evaluate any operators on the other side (or stop executing them if they started already) since you know the join won't produce any rows.

thomasg19930417 · 2022-09-16T01:43:47Z

@ekoifman
Thank you for reply

thomasg19930417 · 2022-09-19T10:22:31Z

@ekoifman hi, when Inner Join and LHS has many empty partition ,why inner join not demote broadcast

   else if (manyEmptyInOther && canBroadcastPlan) {
         join.joinType match {
        case LeftOuter | RightOuter | LeftAnti => true
        case _ => false
      }

…ynamicJoinSelection In the current implementation of DynamicJoinSelection the logic checks if one side of the join has high ratio of empty partitions and adds a NO_BROADCAST hint on that side since a shuffle join can short-circuit the local joins where one side is empty. This logic is doesn't make sense for all join type. For example, a Left Outer Join cannot short circuit if RHS is empty so we should not inhibit BHJ. On the other hand a LOJ executed as a shuffle join where the LHS has many empty can short circuit the local join so we should inhibit the BHJ because BHJ will use OptimizeShuffleWithLocalRead which will re-assemble LHS partitions as the were before the shuffle and thus may not have many empty ones any more. This supersedes [SPARK-37193](https://issues.apache.org/jira/browse/SPARK-37193) Also see previous discussion in #34464 (comment) It may change which joins run as BHJ vs shuffle joins Unit Tests Closes #35715 from ekoifman/SPARK-37753-enhance-DynamiJoinSelection. Authored-by: Eugene Koifman <eugene.koifman@workday.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>

github-actions bot added the SQL label Nov 2, 2021

HyukjinKwon reviewed Nov 2, 2021

View reviewed changes

sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala Outdated Show resolved Hide resolved

singhpk234 reviewed Nov 2, 2021

View reviewed changes

ekoifman force-pushed the SPARK-37193 branch 3 times, most recently from dd11a1e to 08551fb Compare November 9, 2021 18:06

github-actions bot added CORE PYTHON labels Nov 10, 2021

ulysses-you reviewed Nov 16, 2021

View reviewed changes

cloud-fan approved these changes Dec 22, 2021

View reviewed changes

[SPARK-37193][SQL] DynamicJoinSelection.shouldDemoteBroadcastHashJoin…

050f745

… should not apply to outer joins

ekoifman force-pushed the SPARK-37193 branch from 8a3f27d to 050f745 Compare December 22, 2021 15:15

cloud-fan closed this in 66c2d1f Dec 23, 2021

cloud-fan reviewed Dec 23, 2021

View reviewed changes

ekoifman mentioned this pull request Mar 3, 2022

[SPARK-37753][SQL] Fine tune logic to demote Broadcast hash join in DynamicJoinSelection #35715

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-37193][SQL] DynamicJoinSelection.shouldDemoteBroadcastHashJoin should not apply to outer joins #34464

[SPARK-37193][SQL] DynamicJoinSelection.shouldDemoteBroadcastHashJoin should not apply to outer joins #34464

ekoifman commented Nov 2, 2021

AmplabJenkins commented Nov 2, 2021

singhpk234 Nov 2, 2021 •

edited

ekoifman Nov 3, 2021

singhpk234 Nov 2, 2021

ekoifman Nov 2, 2021 •

edited

ekoifman commented Nov 10, 2021

ekoifman commented Nov 15, 2021

ulysses-you Nov 16, 2021

ekoifman Nov 16, 2021 •

edited

ekoifman commented Dec 21, 2021

cloud-fan commented Dec 22, 2021

ekoifman commented Dec 22, 2021

cloud-fan commented Dec 23, 2021

cloud-fan Dec 23, 2021

ekoifman Dec 24, 2021

cloud-fan Dec 27, 2021

ekoifman Dec 27, 2021

cloud-fan commented Jan 7, 2022

thomasg19930417 commented Sep 15, 2022

ekoifman commented Sep 15, 2022

thomasg19930417 commented Sep 16, 2022

thomasg19930417 commented Sep 19, 2022 •

edited

[SPARK-37193][SQL] DynamicJoinSelection.shouldDemoteBroadcastHashJoin should not apply to outer joins #34464

[SPARK-37193][SQL] DynamicJoinSelection.shouldDemoteBroadcastHashJoin should not apply to outer joins #34464

Conversation

ekoifman commented Nov 2, 2021

What changes were proposed in this pull request?

Does this PR introduce any user-facing change?

How was this patch tested?

AmplabJenkins commented Nov 2, 2021

singhpk234 Nov 2, 2021 • edited

Choose a reason for hiding this comment

ekoifman Nov 3, 2021

Choose a reason for hiding this comment

singhpk234 Nov 2, 2021

Choose a reason for hiding this comment

ekoifman Nov 2, 2021 • edited

Choose a reason for hiding this comment

ekoifman commented Nov 10, 2021

ekoifman commented Nov 15, 2021

ulysses-you Nov 16, 2021

Choose a reason for hiding this comment

ekoifman Nov 16, 2021 • edited

Choose a reason for hiding this comment

ekoifman commented Dec 21, 2021

cloud-fan commented Dec 22, 2021

ekoifman commented Dec 22, 2021

cloud-fan commented Dec 23, 2021

cloud-fan Dec 23, 2021

Choose a reason for hiding this comment

ekoifman Dec 24, 2021

Choose a reason for hiding this comment

cloud-fan Dec 27, 2021

Choose a reason for hiding this comment

ekoifman Dec 27, 2021

Choose a reason for hiding this comment

cloud-fan commented Jan 7, 2022

thomasg19930417 commented Sep 15, 2022

ekoifman commented Sep 15, 2022

thomasg19930417 commented Sep 16, 2022

thomasg19930417 commented Sep 19, 2022 • edited

singhpk234 Nov 2, 2021 •

edited

ekoifman Nov 2, 2021 •

edited

ekoifman Nov 16, 2021 •

edited

thomasg19930417 commented Sep 19, 2022 •

edited