[SPARK-37597][WIP] Deduplicate the right side of left-semi join and left-anti join

[zhengruifeng]

What changes were proposed in this pull request?

Deduplicate the right side of left-semi join and left-anti join

Why are the changes needed?

1, reduce the shuffle amount in the right side;
2, improve the chance to broadcast the right side;
3, reslove skewed keys in the right side;

Does this PR introduce any user-facing change?

No

How was this patch tested?

added testsuits

[wangyum]

It should after RewriteSubquery to cover more cases.

[zhengruifeng]

test case:

spark.range(0, 10000, 1, 10).selectExpr("id % 1000 as key1", "id % 3 as value1").createOrReplaceTempView("table1")

spark.range(0, 10000000, 1, 10).selectExpr("id % 100 as key2", "id as value2").createOrReplaceTempView("table2")

spark.sql("SELECT key1 FROM table1 LEFT ANTI JOIN table2 ON key1 = key2").write.mode("overwrite").parquet("/tmp/tmp1.parquet")

master:

== Physical Plan ==
Execute InsertIntoHadoopFsRelationCommand (22)
+- AdaptiveSparkPlan (21)
   +- == Final Plan ==
      * Project (14)
      +- * SortMergeJoin LeftAnti (13)
         :- * Sort (5)
         :  +- AQEShuffleRead (4)
         :     +- ShuffleQueryStage (3)
         :        +- Exchange (2)
         :           +- * Range (1)
         +- * Sort (12)
            +- AQEShuffleRead (11)
               +- ShuffleQueryStage (10)
                  +- Exchange (9)
                     +- * Project (8)
                        +- * Filter (7)
                           +- * Range (6)

this pr:

== Physical Plan ==
Execute InsertIntoHadoopFsRelationCommand (25)
+- AdaptiveSparkPlan (24)
   +- == Final Plan ==
      * Project (16)
      +- * BroadcastHashJoin LeftAnti BuildRight (15)
         :- AQEShuffleRead (4)
         :  +- ShuffleQueryStage (3)
         :     +- Exchange (2)
         :        +- * Range (1)
         +- BroadcastQueryStage (14)
            +- BroadcastExchange (13)
               +- * HashAggregate (12)
                  +- AQEShuffleRead (11)
                     +- ShuffleQueryStage (10)
                        +- Exchange (9)
                           +- * HashAggregate (8)
                              +- * Project (7)
                                 +- * Filter (6)
                                    +- * Range (5)

[wangyum]

Deduplicate is not always has benefit. This is my initial PR: b559901

[zhengruifeng]

yes, there are some cases that had not been taken into account in this PR. I think b559901 is much better. I will close this one.

[SparkQA]

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

[SparkQA]

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

[SparkQA]

Test build #146038 has finished for PR 34850 at commit 49dd1ad.

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