[SPARK-32129][SQL] Support AQE skew join with Union

[LantaoJin]

What changes were proposed in this pull request?

In the apply method of OptimizeSkewedJoin, we first match out the UnionExec nodes, then try to optimize their children with current logic.

Why are the changes needed?

Current, the AQE skew join only supports two tables join such as

SMJ
:-Sort
: +-Shuffle
+-Sort
+-Shuffle

But if the plan contains a Union, the skew join handling not work:

Union
:-SMJ
: :-Sort
: : +-Shuffle
: +-Sort
: +-Shuffle
+-SMJ
: :-Sort
: : +-Shuffle
: +-Sort
+-Shuffle

Does this PR introduce any user-facing change?

No

How was this patch tested?

Add a UT.

[LantaoJin]

retest this please

[SparkQA]

Test build #124652 has finished for PR 28947 at commit d011e9a.

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

[LantaoJin]

retest this please

[SparkQA]

Test build #124705 has finished for PR 28947 at commit d011e9a.

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

[LantaoJin]

retest this please

[LantaoJin]

ping @cloud-fan@gatorsmile

[SparkQA]

Test build #124736 has finished for PR 28947 at commit d011e9a.

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

[LantaoJin]

retest this please

[SparkQA]

Test build #124740 has finished for PR 28947 at commit d011e9a.

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

[cloud-fan]

Can we make it more general? It seems like we can optimize any SMJ if its 2 children are both shuffle stages. cc @JkSelf@maryannxue

[cloud-fan]

For 3-table join, if they are in the same query stage, it means the shuffles are all leaf, and we will only optimize the first SMJ, as the second SMJ has only one side as shuffle stage.

[LantaoJin]

For 3-table join, if they are in the same query stage, it means the shuffles are all leaf, and we will only optimize the first SMJ, as the second SMJ has only one side as shuffle stage.

We have a 3-table skewed join implamentation in our internal code base. But we have replaced the skew join handling logic by community's. So our optimization is not work based on currnet OptimizeSkewedJoin. I will try to re-implement it in current OptimizeSkewedJoin and submit a PR later.

[LantaoJin]

Can we make it more general? It seems like we can optimize any SMJ if its 2 children are both shuffle stages. cc @JkSelf@maryannxue

Yes. we usually implemented some optimizations based on our inner usages and issues. So it may be not general. I only see the UNION case so far.

[LantaoJin]

retest this please

[SparkQA]

Test build #124785 has finished for PR 28947 at commit d011e9a.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[cloud-fan]

Can we make a general approach here? e.g. if we just optimize SMJ with both sides as shuffle stages, we can even optimize the first join of a 3-table join plan.

[LantaoJin]

Can we make a general approach here? e.g. if we just optimize SMJ with both sides as shuffle stages, we can even optimize the first join of a 3-table join plan.

#29021 is the PR to handle more general skew pattern includes n-tables join.

[manuzhang]

@cloud-fan@LantaoJin
Any progress or any suggestion on where we can move forward with this improvement ? We've seen a lot of skew joins not being handled by AQE due to union.

cc @maryannxue@JkSelf

[github-actions]

We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!